Spaces:
Sleeping
Sleeping
File size: 2,366 Bytes
0f24635 32c9c7b 0f24635 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 |
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, update
from src.core.database import get_db
from src.models.postgres.user_model import User
from pydantic import BaseModel, Field
from typing import Optional
router = APIRouter()
class UserCreate(BaseModel):
email: str
hashed_password: str
name: Optional[str] = None
picture_url: Optional[str] = None
google_id: Optional[str] = None
candidate_mongo_id: Optional[str] = None
class UserUpdate(BaseModel):
email: Optional[str] = None
hashed_password: Optional[str] = None
name: Optional[str] = None
picture_url: Optional[str] = None
google_id: Optional[str] = None
candidate_mongo_id: Optional[str] = None
class UserResponse(BaseModel):
id: int
email: str
name: Optional[str] = None
picture_url: Optional[str] = None
google_id: Optional[str] = None
candidate_mongo_id: Optional[str] = None
class Config:
from_attributes = True
@router.post("/users", response_model=UserResponse)
async def create_user(user: UserCreate, db: AsyncSession = Depends(get_db)):
db_user = User(**user.model_dump())
db.add(db_user)
await db.commit()
await db.refresh(db_user)
return db_user
@router.get("/users/{user_id}", response_model=UserResponse)
async def get_user_by_id(user_id: int, db: AsyncSession = Depends(get_db)):
result = await db.execute(select(User).where(User.id == user_id))
user = result.scalar_one_or_none()
if user is None:
raise HTTPException(status_code=404, detail="User not found")
return user
@router.get("/users/email/{email}", response_model=UserResponse)
async def get_user_by_email(email: str, db: AsyncSession = Depends(get_db)):
result = await db.execute(select(User).where(User.email == email))
user = result.scalar_one_or_none()
if user is None:
raise HTTPException(status_code=404, detail="User not found")
return user
@router.put("/users/{user_id}", response_model=UserResponse)
async def update_user(user_id: int, user: UserUpdate, db: AsyncSession = Depends(get_db)):
query = update(User).where(User.id == user_id).values(**user.model_dump(exclude_unset=True))
await db.execute(query)
await db.commit()
return await get_user_by_id(user_id, db)
|