68 lines
2.0 KiB
Python
68 lines
2.0 KiB
Python
|
"""This module contains endpoints for admin operations."""
|
||
|
|
|
||
|
|
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.schemas import common as commonschema
|
||
|
|
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
|
||
|
|
|
||
|
|
|
||
|
|
auth_handler = auth.AuthHandler()
|
||
|
|
|
||
|
|
router = APIRouter(
|
||
|
|
prefix="/admin",
|
||
|
|
tags=["admin"],
|
||
|
|
dependencies=[Depends(auth_handler.asset_current_user_is_admin)],
|
||
|
|
)
|
||
|
|
|
||
|
|
tag_metadata = {
|
||
|
|
"name": "admin",
|
||
|
|
"description": "Operations for administrators."
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
@router.post("/users/", 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
|
||
|
|
|
||
|
|
return usercrud.create_user(db=db, user=user)
|
||
|
|
|
||
|
|
|
||
|
|
@router.get("/users/", response_model=list[userschema.User])
|
||
|
|
def read_users(
|
||
|
|
commons: Annotated[UserSortablePaginationParams, Depends(UserSortablePaginationParams)],
|
||
|
|
db: Session = Depends(get_db),
|
||
|
|
):
|
||
|
|
try:
|
||
|
|
return usercrud.read_users(
|
||
|
|
db=db,
|
||
|
|
skip=commons.skip,
|
||
|
|
limit=commons.limit,
|
||
|
|
sortby=commons.sortby,
|
||
|
|
sortorder=commons.sortorder
|
||
|
|
)
|
||
|
|
except InvalidFilterParameterException as e:
|
||
|
|
raise HTTPException(400, fmt(str(e)))
|
||
|
|
|
||
|
|
|
||
|
|
@router.get("/users/total/", response_model=commonschema.ItemCount)
|
||
|
|
def read_users_count(db: Session = Depends(get_db)):
|
||
|
|
|
||
|
|
return commonschema.ItemCount(total=usercrud.read_users_count(db=db))
|