fastapi-svelte-template/backend/todo/routes/users.py

"""This module contains endpoints for operations related to users."""

from typing import Annotated

from fastapi import APIRouter, Depends, HTTPException, Request
from sqlalchemy.orm import Session

from todo.database.engine import get_db
from todo.schemas import users as userschema
from todo.crud import users as usercrud
from todo.utils.exceptions import NotFoundException, InvalidFilterParameterException
from todo.utils.exceptions import create_exception_dict as fmt
from todo.dependencies.users import UserSortablePaginationParams
import todo.auth.auth as auth

import logging
logger = logging.getLogger()


router = APIRouter(
    prefix="/users",
    tags=["users"]
)

tag_metadata = {
    "name": "users",
    "description": "Operations related to users."
}

auth_handler = auth.AuthHandler()


@router.post("/", response_model=userschema.User)
def create_user(
        user: userschema.UserCreate,
        db: Session = Depends(get_db),
    ):
    # Check if user already exists
    try:
        usercrud.read_user_by_email(db, email=user.email)
        raise HTTPException(400, "A user with this email address is already registered.")
    except NotFoundException:
        pass

    if user.is_admin:
        raise HTTPException(403, "You are not authorized to perform this action.")

    return usercrud.create_user(db=db, user=user)


@router.get("/{user_id}", response_model=userschema.User)
def read_user(
        user_id: int,
        request: Request,
        db: Session = Depends(get_db),
        current_user: userschema.User = Depends(auth_handler.get_current_user),
    ):
    logger.error(request)
    try:
        user = usercrud.read_user(db=db, id=user_id)
    except NotFoundException as e:
        raise HTTPException(404, fmt(str(e)))

    if current_user.is_admin or current_user.id == user_id:
        return user
    raise HTTPException(403, "You are not authorized to view this content.")


@router.patch("/{id}", response_model=userschema.User)
def update_user(
        user_id: int,
        user: userschema.UserUpdate,
        db: Session = Depends(get_db),
        current_user: userschema.User = Depends(auth_handler.get_current_user),
    ):
    if not (current_user.is_admin or current_user.id == user_id):
        raise HTTPException(403, "You are not authorized to perform this action.")

    if user.is_admin and not current_user.is_admin:
        raise HTTPException(403, "You are not authorized to perform this action.")

    try:
        return usercrud.update_user(db=db, user=user, id=user_id)
    except NotFoundException as e:
        raise HTTPException(404, fmt(str(e)))


@router.delete("/{id}", response_model=userschema.User)
def delete_user(
        user_id: int,
        db: Session = Depends(get_db),
        current_user: userschema.User = Depends(auth_handler.get_current_user),
    ):
    if not (current_user.is_admin or current_user.id == user_id):
        raise HTTPException(403, "You are not authorized to perform this action.")

    try:
        return usercrud.delete_user(db=db, id=user_id)
    except NotFoundException as e:
        raise HTTPException(404, fmt(str(e)))