95 lines
3 KiB
Python
95 lines
3 KiB
Python
"""This module contains endpoints for operations related to users."""
|
|
|
|
from typing import Annotated
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
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
|
|
|
|
|
|
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,
|
|
db: Session = Depends(get_db),
|
|
current_user: userschema.User = Depends(auth_handler.get_current_user),
|
|
):
|
|
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)))
|