File size: 4,580 Bytes
d99646e 39a56a6 d99646e 39a56a6 d99646e 39a56a6 d99646e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 | from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from app.dependencies import get_current_admin_user, get_db
from app.models.user import User
from app.schemas.admin import AdminDepartmentUpdate, AdminUserUpdate
from app.schemas.department import DepartmentCreate, DepartmentResponse
from app.schemas.query import QueryRead
from app.schemas.user import UserRead
from app.services.admin_service import (
create_department,
deactivate_department,
deactivate_user,
get_department,
get_user,
list_departments,
list_users,
update_department,
update_user,
)
from app.services.query_service import list_all_queries
router = APIRouter(prefix="/admin", tags=["admin"])
def _raise_not_found(entity: str) -> None:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"{entity} not found",
)
@router.get("/departments", response_model=list[DepartmentResponse])
def admin_list_departments(
db: Session = Depends(get_db),
current_admin: User = Depends(get_current_admin_user),
):
_ = current_admin
return list_departments(db)
@router.post("/departments", response_model=DepartmentResponse, status_code=status.HTTP_201_CREATED)
def admin_create_department(
payload: DepartmentCreate,
db: Session = Depends(get_db),
current_admin: User = Depends(get_current_admin_user),
):
_ = current_admin
try:
return create_department(db, **payload.model_dump())
except ValueError as exc:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)) from exc
@router.put("/departments/{department_id}", response_model=DepartmentResponse)
def admin_update_department(
department_id: int,
payload: AdminDepartmentUpdate,
db: Session = Depends(get_db),
current_admin: User = Depends(get_current_admin_user),
):
_ = current_admin
try:
department = update_department(db, department_id=department_id, payload=payload)
except ValueError as exc:
message = str(exc)
status_code = status.HTTP_404_NOT_FOUND if message == "Department not found" else status.HTTP_400_BAD_REQUEST
raise HTTPException(status_code=status_code, detail=message) from exc
return department
@router.delete("/departments/{department_id}", response_model=DepartmentResponse)
def admin_deactivate_department(
department_id: int,
db: Session = Depends(get_db),
current_admin: User = Depends(get_current_admin_user),
):
_ = current_admin
try:
department = deactivate_department(db, department_id=department_id)
except ValueError as exc:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc
return department
@router.get("/users", response_model=list[UserRead])
def admin_list_users(
db: Session = Depends(get_db),
current_admin: User = Depends(get_current_admin_user),
):
_ = current_admin
return list_users(db)
@router.get("/queries", response_model=list[QueryRead])
def admin_list_queries(
db: Session = Depends(get_db),
current_admin: User = Depends(get_current_admin_user),
):
_ = current_admin
return list_all_queries(db)
@router.get("/users/{user_id}", response_model=UserRead)
def admin_get_user(
user_id: int,
db: Session = Depends(get_db),
current_admin: User = Depends(get_current_admin_user),
):
_ = current_admin
user = get_user(db, user_id=user_id)
if user is None:
_raise_not_found("User")
return user
@router.put("/users/{user_id}", response_model=UserRead)
def admin_update_user(
user_id: int,
payload: AdminUserUpdate,
db: Session = Depends(get_db),
current_admin: User = Depends(get_current_admin_user),
):
_ = current_admin
try:
user = update_user(db, user_id=user_id, payload=payload)
except ValueError as exc:
message = str(exc)
status_code = status.HTTP_404_NOT_FOUND if message == "User not found" else status.HTTP_400_BAD_REQUEST
raise HTTPException(status_code=status_code, detail=message) from exc
return user
@router.delete("/users/{user_id}", response_model=UserRead)
def admin_deactivate_user(
user_id: int,
db: Session = Depends(get_db),
current_admin: User = Depends(get_current_admin_user),
):
_ = current_admin
try:
user = deactivate_user(db, user_id=user_id)
except ValueError as exc:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc
return user
|