from uuid import UUID from fastapi import APIRouter, HTTPException from fastapi.params import Depends from sqlalchemy.orm import Session as DbSession from app.database import get_db from app.schemas.session import Session from app.schemas.auth_token import AccessToken from app.schemas.user import AdministrativeUserUpdate, User, UserCreate from app.security.jwt_bearer import JWTBearer from app.services import session_service, user_service from app.util.errors import NotFoundError router = APIRouter( prefix="/users", tags=["User (v1)"], ) @router.get(path="", response_model=list[User]) async def get_users( email: str = None, skip: int = 0, limit: int = 20, db: DbSession = Depends(get_db), token: AccessToken = Depends(JWTBearer(required_roles=["administrator"])), ): return await user_service.get_users(db, skip, limit, email) @router.post(path="", status_code=201, response_model=User) async def create_user( create_user: UserCreate, db: DbSession = Depends(get_db), token: AccessToken = Depends(JWTBearer(required_roles=["administrator"])), ): user = await user_service.create_user( db=db, user=create_user ) return user @router.patch(path="/{user_id}", response_model=User) async def update_user( user_id: UUID, user_update: AdministrativeUserUpdate, db: DbSession = Depends(get_db), token: AccessToken = Depends(JWTBearer(required_roles=["administrator"])), ): try: return await user_service.update_user( db, user_id, user_update ) except NotFoundError: raise HTTPException(status_code=404, detail="user_not_found") @router.delete(path="/{user_id}", response_model=None) async def delete_user( user_id: UUID, db: DbSession = Depends(get_db), token: AccessToken = Depends(JWTBearer(required_roles=["administrator"])), ): try: await user_service.remove_user(db, user_id) return list() except NotFoundError: raise HTTPException(status_code=404, detail="user_not_found") @router.get( path="/{user_id}/sessions", response_model=list[Session] ) async def get_user_sessions( user_id: UUID, db: DbSession = Depends(get_db), token: AccessToken = Depends(JWTBearer(required_roles=["administrator"])), ): """ Query sessions of the specified user. Requires the "administrator" role. """ return await session_service.get_sessions_by_user(db=db, user_id=user_id) @router.delete( path="/{user_id}/sessions", response_model=list[None] ) async def remove_all_user_session( user_id: UUID, db: DbSession = Depends(get_db), token: AccessToken = Depends(JWTBearer(required_roles=["administrator"])), ): """ Delete all sessions of the specified user. Requires the "administrator" role. """ await session_service.remove_all_sessions_for_user( db=db, user_id=user_id ) return list() @router.delete( path="/{user_id}/sessions/{session_id}", response_model=list[None], ) async def remove_user_session( user_id: UUID, session_id: UUID, db: DbSession = Depends(get_db), token: AccessToken = Depends(JWTBearer(required_roles=["administrator"])), ): """ Delete the specified session of the specified user. Requires the "administrator" role. """ try: await session_service.remove_session_for_user( db=db, id=session_id, user_id=user_id ) except NotFoundError: raise HTTPException(status_code=404, detail="session_not_found") return list()