from typing import Any, List from fastapi import APIRouter, Body, Depends, HTTPException, status from fastapi.encoders import jsonable_encoder from sqlalchemy.orm import Session from app import schemas from app.api.deps import get_current_active_superuser, get_current_active_user, get_db from app.services import user_service router = APIRouter() @router.get("/", response_model=List[schemas.User]) def read_users( db: Session = Depends(get_db), skip: int = 0, limit: int = 100, current_user: schemas.User = Depends(get_current_active_superuser), ) -> Any: """ Retrieve users. """ users = user_service.get_multi(db, skip=skip, limit=limit) return users @router.get("/me", response_model=schemas.User) def read_user_me( db: Session = Depends(get_db), current_user: schemas.User = Depends(get_current_active_user), ) -> Any: """ Get current user. """ return current_user @router.patch("/me", response_model=schemas.User) def update_user_me( *, db: Session = Depends(get_db), full_name: str = Body(None), email: str = Body(None), password: str = Body(None), current_user: schemas.User = Depends(get_current_active_user), ) -> Any: """ Update own user. """ current_user_data = jsonable_encoder(current_user) user_in = schemas.UserUpdate(**current_user_data) if full_name is not None: user_in.full_name = full_name if email is not None: user_in.email = email if password is not None: user_in.password = password user = user_service.update(db, db_obj=current_user, obj_in=user_in) return user @router.get("/{user_id}", response_model=schemas.User) def read_user_by_id( user_id: int, current_user: schemas.User = Depends(get_current_active_user), db: Session = Depends(get_db), ) -> Any: """ Get a specific user by id. """ user = user_service.get(db, id=user_id) if user == current_user: return user if not current_user.is_superuser: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="The user doesn't have enough privileges" ) return user