Spaces:
Sleeping
Sleeping
File size: 2,753 Bytes
5753020 d5684b3 5753020 d5684b3 5753020 d5684b3 5753020 d5684b3 5753020 d5684b3 5753020 d5684b3 5753020 d5684b3 5753020 d5684b3 5753020 d5684b3 5753020 d5684b3 5753020 d5684b3 5753020 d5684b3 | 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 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 | import logging
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from app.engine.postgresdb import get_db_session
from app.schema.index import UserCreate, User as UserSchema
from app.model.user import User as UserModel
user_router = r = APIRouter(prefix="/api/v1/users", tags=["users"])
logger = logging.getLogger(__name__)
@r.post(
"/",
response_model=UserSchema,
responses={
200: {"description": "New user created"},
400: {"description": "Bad request"},
409: {"description": "Conflict"},
500: {"description": "Internal server error"},
},
)
async def create_user(user: UserCreate, db: AsyncSession = Depends(get_db_session)):
try:
logger.info(f"Checking if user exists: {user.dict()}")
db_user = await UserModel.get(db, email=user.email)
if db_user and not db_user.is_deleted:
raise HTTPException(status_code=409, detail="User already exists")
logger.info(f"Creating user: {user.dict()}")
db_user = await UserModel.create(db, **user.dict())
return db_user
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@r.get(
"/{user_id}",
response_model=UserSchema,
responses={
200: {"description": "User found"},
404: {"description": "User not found"},
500: {"description": "Internal server error"},
},
)
async def get_user(user_id: int, db: AsyncSession = Depends(get_db_session)):
user = await UserModel.get(db, id=user_id)
if not user:
raise HTTPException(status_code=404, detail="User not found")
return user
@r.put(
"/{user_id}",
response_model=UserSchema,
responses={
200: {"description": "User updated"},
404: {"description": "User not found"},
500: {"description": "Internal server error"},
},
)
async def update_user(user_id: int, user_payload: UserCreate, db: AsyncSession = Depends(get_db_session)):
user = await UserModel.get(db, id=user_id)
if not user:
raise HTTPException(status_code=404, detail="User not found")
user = await UserModel.update(db, id=user_id, **user_payload.dict())
return user
@r.delete(
"/{user_id}",
response_model=UserSchema,
responses={
200: {"description": "User deleted"},
404: {"description": "User not found"},
500: {"description": "Internal server error"},
},
)
async def delete_user(user_id: int, db: AsyncSession = Depends(get_db_session)):
user = await UserModel.get(db, id=user_id)
if not user:
raise HTTPException(status_code=404, detail="User not found")
user = await UserModel.delete(db, id=user_id)
return user
|