""" User management endpoints """ from typing import Any, List from app import crud, models, schemas from app.api import deps from fastapi import APIRouter, Depends, HTTPException, status from sqlalchemy.orm import Session router = APIRouter() @router.get("", response_model=List[schemas.User]) def read_users( db: Session = Depends(deps.get_db), skip: int = 0, limit: int = 100, current_user: models.User = Depends(deps.get_current_active_superuser), ) -> Any: """ Retrieve users (superuser only) """ users = crud.user.get_multi(db, skip=skip, limit=limit) return users @router.get("/{user_id}", response_model=schemas.User) def read_user( user_id: int, current_user: models.User = Depends(deps.get_current_active_user), db: Session = Depends(deps.get_db), ) -> Any: """ Get a specific user by id """ user = crud.user.get(db, id=user_id) if user == current_user: return user if not crud.user.is_superuser(current_user): raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="The user doesn't have enough privileges" ) if not user: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="User not found" ) return user @router.put("/me", response_model=schemas.User) def update_user_me( *, db: Session = Depends(deps.get_db), user_in: schemas.UserUpdate, current_user: models.User = Depends(deps.get_current_active_user), ) -> Any: """ Update own user """ if user_in.username is not None: user_with_username = crud.user.get_by_username(db, username=user_in.username) if user_with_username and user_with_username.id != current_user.id: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Username already registered" ) if user_in.email is not None: user_with_email = crud.user.get_by_email(db, email=user_in.email) if user_with_email and user_with_email.id != current_user.id: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Email already registered" ) user = crud.user.update(db, db_obj=current_user, obj_in=user_in) return user