Moved role.py validation logic to request models and used custom HTTP exceptions

Author: chriscarrollsmith

routers/role.py

@@ -1,9 +1,9 @@
 from typing import List
 from datetime import datetime
 from logging import getLogger
-from fastapi import APIRouter, Depends, HTTPException, Form
+from fastapi import APIRouter, Depends, Form, HTTPException
 from fastapi.responses import RedirectResponse
-from pydantic import BaseModel, ConfigDict
+from pydantic import BaseModel, ConfigDict, field_validator
 from sqlmodel import Session, select
 from utils.db import get_session
 from utils.models import Role, RolePermissionLink, ValidPermissions, utc_time
@@ -13,15 +13,52 @@
 router = APIRouter(prefix="/roles", tags=["roles"])
 
 
+# -- Custom Exceptions --
+
+class RoleAlreadyExistsError(HTTPException):
+    """Raised when attempting to create a role with a name that already exists"""
+
+    def __init__(self):
+        super().__init__(status_code=400, detail="Role already exists")
+
+
+class RoleNotFoundError(HTTPException):
+    """Raised when a requested role does not exist or is deleted"""
+
+    def __init__(self):
+        super().__init__(status_code=404, detail="Role not found")
+
+
+# -- Server Request Models --
+
 class RoleCreate(BaseModel):
     model_config = ConfigDict(from_attributes=True)
 
     name: str
     permissions: List[ValidPermissions]
 
+    @field_validator("name")
+    @classmethod
+    def validate_unique_name(cls, name: str, info):
+        # Note: This requires passing session as a dependency to as_form
+        session = info.context.get("session")
+        if session and session.exec(select(Role).where(Role.name == name)).first():
+            raise RoleAlreadyExistsError()
+        return name
+
     @classmethod
-    async def as_form(cls, name: str = Form(...), permissions: List[ValidPermissions] = Form(...)):
-        return cls(name=name, permissions=permissions)
+    async def as_form(
+        cls,
+        name: str = Form(...),
+        permissions: List[ValidPermissions] = Form(...),
+        session: Session = Depends(get_session)
+    ):
+        # Pass session to validator context
+        return cls(
+            name=name,
+            permissions=permissions,
+            context={"session": session}
+        )
 
 
 class RoleRead(BaseModel):
@@ -42,20 +79,39 @@ class RoleUpdate(BaseModel):
     name: str
     permissions: List[ValidPermissions]
 
+    @field_validator("id")
+    @classmethod
+    def validate_role_exists(cls, id: int, info):
+        session = info.context.get("session")
+        if session:
+            role = session.get(Role, id)
+            if not role or not role.id or role.deleted:
+                raise RoleNotFoundError()
+        return id
+
     @classmethod
-    async def as_form(cls, id: int = Form(...), name: str = Form(...), permissions: List[ValidPermissions] = Form(...)):
-        return cls(id=id, name=name, permissions=permissions)
+    async def as_form(
+        cls,
+        id: int = Form(...),
+        name: str = Form(...),
+        permissions: List[ValidPermissions] = Form(...),
+        session: Session = Depends(get_session)
+    ):
+        return cls(
+            id=id,
+            name=name,
+            permissions=permissions,
+            context={"session": session}
+        )
+
 
+# -- Routes --
 
 @router.post("/", response_class=RedirectResponse)
 def create_role(
     role: RoleCreate = Depends(RoleCreate.as_form),
     session: Session = Depends(get_session)
 ) -> RedirectResponse:
-    db_role = session.exec(select(Role).where(Role.name == role.name)).first()
-    if db_role:
-        raise HTTPException(status_code=400, detail="Role already exists")
-
     # Create role and permissions in a single transaction
     db_role = Role(name=role.name)
 
@@ -66,7 +122,7 @@ def create_role(
     ]
 
     session.add(db_role)
-    session.commit()  # Commit once after all operations
+    session.commit()
 
     return RedirectResponse(url="/roles", status_code=303)
 
@@ -75,7 +131,7 @@ def create_role(
 def read_role(role_id: int, session: Session = Depends(get_session)):
     db_role: Role | None = session.get(Role, role_id)
     if not db_role or not db_role.id or db_role.deleted:
-        raise HTTPException(status_code=404, detail="Role not found")
+        raise RoleNotFoundError()
 
     permissions = [
         ValidPermissions(link.permission.name)
@@ -99,8 +155,7 @@ def update_role(
     session: Session = Depends(get_session)
 ) -> RedirectResponse:
     db_role: Role | None = session.get(Role, role.id)
-    if not db_role or not db_role.id or db_role.deleted:
-        raise HTTPException(status_code=404, detail="Role not found")
+
     role_data = role.model_dump(exclude_unset=True)
     for key, value in role_data.items():
         setattr(db_role, key, value)
@@ -130,7 +185,8 @@ def delete_role(
 ) -> RedirectResponse:
     db_role = session.get(Role, role_id)
     if not db_role:
-        raise HTTPException(status_code=404, detail="Role not found")
+        raise RoleNotFoundError()
+
     db_role.deleted = True
     db_role.updated_at = utc_time()
     session.add(db_role)