60 lines
1.9 KiB
Python
60 lines
1.9 KiB
Python
|
"""This module contains endpoints for operations related to users."""
|
||
|
|
|
||
|
|
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.exceptions import NotFoundException
|
||
|
|
|
||
|
|
|
||
|
|
router = APIRouter(
|
||
|
|
prefix="/users",
|
||
|
|
tags=["users"]
|
||
|
|
)
|
||
|
|
|
||
|
|
tag_metadata = {
|
||
|
|
"name": "users",
|
||
|
|
"description": "Operations related to users."
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
@router.post("/", response_model=userschema.User)
|
||
|
|
def create_user(user: userschema.UserCreate, db: Session = Depends(get_db)):
|
||
|
|
try:
|
||
|
|
# An exception is expected here, because we need to check if a user with this email is already registered
|
||
|
|
usercrud.read_user_by_email(db, email=user.email)
|
||
|
|
raise HTTPException(400, "A user with this email address is already registered.")
|
||
|
|
except NotFoundException:
|
||
|
|
return usercrud.create_user(db=db, user=user)
|
||
|
|
|
||
|
|
|
||
|
|
@router.get("/{id}", response_model=userschema.User)
|
||
|
|
def read_user(id: int, db: Session = Depends(get_db)):
|
||
|
|
try:
|
||
|
|
return usercrud.read_user(db=db, id=id)
|
||
|
|
except NotFoundException as e:
|
||
|
|
raise HTTPException(404, str(e))
|
||
|
|
|
||
|
|
|
||
|
|
@router.get("/", response_model=list[userschema.User])
|
||
|
|
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
|
||
|
|
return usercrud.read_users(db=db, skip=skip, limit=limit)
|
||
|
|
|
||
|
|
|
||
|
|
@router.patch("/{id}", response_model=userschema.User)
|
||
|
|
def update_user(id: int, user: userschema.UserUpdate, db: Session = Depends(get_db)):
|
||
|
|
try:
|
||
|
|
return usercrud.update_user(db=db, user=user, id=id)
|
||
|
|
except NotFoundException as e:
|
||
|
|
raise HTTPException(404, str(e))
|
||
|
|
|
||
|
|
|
||
|
|
@router.delete("/{id}", response_model=userschema.User)
|
||
|
|
def delete_user(id: int, db: Session = Depends(get_db)):
|
||
|
|
try:
|
||
|
|
return usercrud.delete_user(db=db, id=id)
|
||
|
|
except NotFoundException as e:
|
||
|
|
raise HTTPException(404, str(e))
|