from typing import Any, List from fastapi import APIRouter, Depends, HTTPException, status from sqlalchemy.orm import Session from app import crud, models, schemas from app.api.v1 import deps router = APIRouter() @router.get("/", response_model=List[schemas.Role]) def read_roles( db: Session = Depends(deps.get_db), skip: int = 0, limit: int = 100, current_user: models.User = Depends(deps.get_current_active_user), ) -> Any: """ Retrieve roles. """ roles = crud.role.get_multi(db, skip=skip, limit=limit) return roles @router.post("/", response_model=schemas.Role) def create_role( *, db: Session = Depends(deps.get_db), role_in: schemas.RoleCreate, current_user: models.User = Depends(deps.get_admin_user), ) -> Any: """ Create new role. """ role = crud.role.get_by_name(db, name=role_in.name) if role: raise HTTPException( status_code=400, detail="The role with this name already exists in the system.", ) role = crud.role.create(db, obj_in=role_in) return role @router.get("/{role_id}", response_model=schemas.Role) def read_role( *, db: Session = Depends(deps.get_db), role_id: int, current_user: models.User = Depends(deps.get_current_active_user), ) -> Any: """ Get role by ID. """ role = crud.role.get(db, id=role_id) if not role: raise HTTPException( status_code=404, detail="The role with this ID does not exist in the system", ) return role @router.put("/{role_id}", response_model=schemas.Role) def update_role( *, db: Session = Depends(deps.get_db), role_id: int, role_in: schemas.RoleUpdate, current_user: models.User = Depends(deps.get_admin_user), ) -> Any: """ Update a role. """ role = crud.role.get(db, id=role_id) if not role: raise HTTPException( status_code=404, detail="The role with this ID does not exist in the system", ) # Check if trying to update to an existing name if role_in.name and role_in.name != role.name: existing_role = crud.role.get_by_name(db, name=role_in.name) if existing_role: raise HTTPException( status_code=400, detail="The role with this name already exists in the system.", ) role = crud.role.update(db, db_obj=role, obj_in=role_in) return role @router.delete("/{role_id}", status_code=status.HTTP_204_NO_CONTENT, response_model=None) def delete_role( *, db: Session = Depends(deps.get_db), role_id: int, current_user: models.User = Depends(deps.get_admin_user), ) -> Any: """ Delete a role. """ role = crud.role.get(db, id=role_id) if not role: raise HTTPException( status_code=404, detail="The role with this ID does not exist in the system", ) # Check if there are users with this role users_with_role = db.query(models.User).filter(models.User.role_id == role_id).count() if users_with_role > 0: raise HTTPException( status_code=400, detail=f"Cannot delete role with ID {role_id}. There are {users_with_role} users with this role.", ) crud.role.remove(db, id=role_id) return None