Spaces:
Sleeping
Sleeping
File size: 3,373 Bytes
f92858b d5684b3 f92858b d5684b3 f92858b 81ec8d6 d5684b3 107b48d d5684b3 f92858b d5684b3 81ec8d6 d5684b3 f92858b d5684b3 f92858b 417abff d5684b3 4761dd9 81ec8d6 417abff 107b48d d610e3e 107b48d 417abff 107b48d 81ec8d6 107b48d d5684b3 81ec8d6 d5684b3 81ec8d6 d5684b3 81ec8d6 d5684b3 81ec8d6 4761dd9 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 91 92 93 94 95 96 97 98 99 100 101 102 103 | 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
from app.model.transaction import Transaction as TransactionModel
from tests.utils import get_fake_transactions
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)
print(f"db_user: {db_user}")
if db_user and not db_user.is_deleted:
raise HTTPException(status_code=409, detail="User already exists")
await UserModel.create(db, **user.model_dump())
user = await UserModel.get(db, email=user.email)
print(f"user: {user}")
# Supply fake transactions data when creating user to have data to query
fake_transactions = get_fake_transactions(user.id)
await TransactionModel.bulk_create(db, fake_transactions)
print(f"fake_transactions: {fake_transactions}")
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.model_dump())
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
|