Spaces:
Sleeping
Sleeping
File size: 2,909 Bytes
f92858b d5684b3 f92858b d5684b3 f92858b 81ec8d6 d5684b3 f92858b d5684b3 81ec8d6 d5684b3 f92858b d5684b3 f92858b d5684b3 81ec8d6 d5684b3 81ec8d6 d5684b3 81ec8d6 d5684b3 81ec8d6 d5684b3 81ec8d6 d5684b3 81ec8d6 d5684b3 81ec8d6 d5684b3 81ec8d6 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 87 88 89 90 | 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, UserResponse, UserUpdate
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=UserResponse,
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:
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")
await UserModel.create(db, **user.dict())
user = await UserModel.get(db, email=user.email)
return user
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@r.get(
"/{email}",
response_model=UserResponse,
responses={
200: {"description": "User found"},
404: {"description": "User not found"},
500: {"description": "Internal server error"},
},
)
async def get_user(email: str, db: AsyncSession = Depends(get_db_session)):
user = await UserModel.get(db, email=email)
if not user:
raise HTTPException(status_code=404, detail="User not found")
return user
@r.put(
"/{email}",
response_model=UserResponse,
responses={
200: {"description": "User updated"},
404: {"description": "User not found"},
500: {"description": "Internal server error"},
},
)
async def update_user(email: str, user_payload: UserUpdate, db: AsyncSession = Depends(get_db_session)):
try:
user = await UserModel.get(db, email=email)
if not user:
raise HTTPException(status_code=404, detail="User not found")
await UserModel.update(db, id=user.id, **user_payload.dict())
user = await UserModel.get(db, email=email)
return user
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@r.delete(
"/{email}",
response_model=UserResponse,
responses={
200: {"description": "User deleted"},
404: {"description": "User not found"},
500: {"description": "Internal server error"},
},
)
async def delete_user(email: str, db: AsyncSession = Depends(get_db_session)):
user = await UserModel.get(db, email=email)
if not user:
raise HTTPException(status_code=404, detail="User not found")
await UserModel.delete(db, email=email)
user = await UserModel.get(db, email=email)
return user
|