1  """Users API endpoints."""
  2  from typing import Any, List
  3  
  4  from fastapi import APIRouter, Depends, HTTPException, status
  5  from sqlalchemy.orm import Session
  6  
  7  from .... import crud, models, schemas
  8  from ....core.security import get_password_hash
  9  from ... import deps
 10  
 11  router = APIRouter()
 12  
 13  
 14  @router.get("/", response_model=List[schemas.User])
 15  def read_users(
 16      db: Session = Depends(deps.get_db),
 17      skip: int = 0,
 18      limit: int = 100,
 19      current_user: models.User = Depends(deps.get_current_active_superuser),
 20  ) -> Any:
 21      """
 22      Retrieve users.
 23      
 24      Only accessible by superusers.
 25      """
 26      users = crud.user.get_multi(db, skip=skip, limit=limit)
 27      return users
 28  
 29  
 30  @router.post("/", response_model=schemas.User, status_code=status.HTTP_201_CREATED)
 31  def create_user(
 32      *,
 33      db: Session = Depends(deps.get_db),
 34      user_in: schemas.UserCreate,
 35      current_user: models.User = Depends(deps.get_current_active_superuser),
 36  ) -> Any:
 37      """
 38      Create new user.
 39      
 40      Only accessible by superusers.
 41      """
 42      user = crud.user.get_by_email(db, email=user_in.email)
 43      if user:
 44          raise HTTPException(
 45              status_code=400,
 46              detail="The user with this email already exists in the system.",
 47          )
 48      user = crud.user.create(db, obj_in=user_in)
 49      return user
 50  
 51  
 52  @router.get("/me", response_model=schemas.User)
 53  def read_user_me(
 54      current_user: models.User = Depends(deps.get_current_active_user),
 55  ) -> Any:
 56      """Get current user."""
 57      return current_user
 58  
 59  
 60  @router.put("/me", response_model=schemas.User)
 61  def update_user_me(
 62      *,
 63      db: Session = Depends(deps.get_db),
 64      user_in: schemas.UserUpdate,
 65      current_user: models.User = Depends(deps.get_current_active_user),
 66  ) -> Any:
 67      """Update own user."""
 68      user = crud.user.update(db, db_obj=current_user, obj_in=user_in)
 69      return user
 70  
 71  
 72  @router.get("/{user_id}", response_model=schemas.User)
 73  def read_user_by_id(
 74      user_id: int,
 75      current_user: models.User = Depends(deps.get_current_active_user),
 76      db: Session = Depends(deps.get_db),
 77  ) -> Any:
 78      """Get a specific user by id."""
 79      user = crud.user.get(db, id=user_id)
 80      if user == current_user:
 81          return user
 82      if not crud.user.is_superuser(current_user):
 83          raise HTTPException(
 84              status_code=400, detail="The user doesn't have enough privileges"
 85          )
 86      return user
 87  
 88  
 89  @router.put("/{user_id}", response_model=schemas.User)
 90  def update_user(
 91      *,
 92      db: Session = Depends(deps.get_db),
 93      user_id: int,
 94      user_in: schemas.UserUpdate,
 95      current_user: models.User = Depends(deps.get_current_active_superuser),
 96  ) -> Any:
 97      """
 98      Update a user.
 99      
100      Only accessible by superusers.
101      """
102      user = crud.user.get(db, id=user_id)
103      if not user:
104          raise HTTPException(
105              status_code=404,
106              detail="The user with this id does not exist in the system",
107          )
108      user = crud.user.update(db, db_obj=user, obj_in=user_in)
109      return user
110  
111  
112  @router.delete("/{user_id}", response_model=schemas.User)
113  def delete_user(
114      *,
115      db: Session = Depends(deps.get_db),
116      user_id: int,
117      current_user: models.User = Depends(deps.get_current_active_superuser),
118  ) -> Any:
119      """
120      Delete a user.
121      
122      Only accessible by superusers.
123      """
124      user = crud.user.get(db, id=user_id)
125      if not user:
126          raise HTTPException(
127              status_code=404,
128              detail="The user with this id does not exist in the system",
129          )
130      user = crud.user.remove(db, id=user_id)
131      return user