Spaces:
Sleeping
Sleeping
File size: 2,719 Bytes
60ef74b | 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 91 92 93 | from typing import List
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from app.engine.postgresdb import get_db
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"])
@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: Session = Depends(get_db)):
try:
db_user = (
db.query(UserModel).filter(UserModel.email == user.email).sort_by(UserModel.updated_at, desc=True).first()
)
if db_user and not db_user.is_deleted:
raise HTTPException(status_code=409, detail="User already exists")
db_user = UserModel(**user.dict())
db.add(db_user)
db.commit()
db.refresh(db_user)
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: Session = Depends(get_db)):
user = db.query(UserModel).get(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: Session = Depends(get_db)):
user = db.query(UserModel).get(user_id)
if not user:
raise HTTPException(status_code=404, detail="User not found")
user.name = user_payload.name
user.email = user_payload.email
db.commit()
db.refresh(user)
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: Session = Depends(get_db)):
user = db.query(UserModel).get(user_id)
if not user:
raise HTTPException(status_code=404, detail="User not found")
user.is_deleted = True
db.commit()
db.refresh(user)
return user
|