69 lines
2.5 KiB
Python
69 lines
2.5 KiB
Python
import uuid
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
|
|
from .models import User, UserCreate, UserRead, UserUpdate
|
|
from .manager import get_user_manager, get_current_user, get_current_superuser
|
|
|
|
from fastapi_pagination import Page
|
|
from fastapi_pagination.ext.sqlmodel import paginate
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
@router.post("", response_description="User added to the database")
|
|
async def create(user_form: UserCreate, user_manager=Depends(get_user_manager), current_user=Depends(get_current_superuser)) -> dict:
|
|
await user_manager.create(user_form, safe=True)
|
|
return {"message": "User added successfully"}
|
|
|
|
|
|
@router.get("", response_model=Page[UserRead], response_description="User records retrieved")
|
|
async def read_list(current_user=Depends(get_current_superuser), user_manager=Depends(get_user_manager)) -> Page[UserRead]:
|
|
return paginate(user_manager.user_db.session, user_manager.user_db.list())
|
|
|
|
|
|
@router.get("/me", response_description="User record retrieved")
|
|
async def read_me(current_user=Depends(get_current_user), user_manager=Depends(get_user_manager)) -> UserRead:
|
|
user = await user_manager.get(current_user.id)
|
|
return user
|
|
|
|
|
|
@router.get("/{user_id}", response_description="User record retrieved")
|
|
async def read_id(user_id: uuid.UUID, current_user=Depends(get_current_superuser), user_manager=Depends(get_user_manager)) -> UserRead:
|
|
user = await user_manager.get(user_id)
|
|
if not user:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail="User not found."
|
|
)
|
|
return UserRead(**user.dict())
|
|
|
|
|
|
@router.put("/{user_id}", response_description="User record updated")
|
|
async def update(user_id: uuid.UUID, user_data: UserUpdate, current_user=Depends(get_current_superuser), user_manager=Depends(get_user_manager)) -> UserRead:
|
|
user = await user_manager.get(user_id)
|
|
if not user:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail="User not found."
|
|
)
|
|
|
|
await user_manager.update(user_data, user, safe=True)
|
|
return user
|
|
|
|
|
|
@router.delete("/{user_id}", response_description="User record deleted from the database")
|
|
async def delete(user_id: uuid.UUID, current_user=Depends(get_current_superuser), user_manager=Depends(get_user_manager)) -> dict:
|
|
user = await user_manager.get(user_id)
|
|
|
|
if not user:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail="User not found."
|
|
)
|
|
|
|
await user_manager.delete(user)
|
|
return {
|
|
"message": "User deleted successfully."
|
|
}
|