Fixed some type lint errors

Author: chriscarrollsmith

routers/organization.py

@@ -1,65 +1,22 @@
 from logging import getLogger
-from datetime import datetime
+from typing import Annotated
 from fastapi import APIRouter, Depends, Form, Request
 from fastapi.responses import RedirectResponse
 from fastapi.templating import Jinja2Templates
-from pydantic import BaseModel, ConfigDict, field_validator
+from pydantic import StringConstraints
 from sqlmodel import Session, select
 from utils.db import get_session, default_roles
 from utils.dependencies import get_authenticated_user, get_user_with_relations
 from utils.models import Organization, User, Role, utc_time
 from utils.enums import ValidPermissions
-from exceptions.http_exceptions import EmptyOrganizationNameError, OrganizationNotFoundError, OrganizationNameTakenError, InsufficientPermissionsError
+from exceptions.http_exceptions import OrganizationNotFoundError, OrganizationNameTakenError, InsufficientPermissionsError
 
 logger = getLogger("uvicorn.error")
 
 router = APIRouter(prefix="/organizations", tags=["organizations"])
 templates = Jinja2Templates(directory="templates")
 
 
-# --- Server Request and Response Models ---
-
-
-class OrganizationCreate(BaseModel):
-    name: str
-
-    @field_validator('name')
-    @classmethod
-    def validate_name(cls, name: str) -> str:
-        if not name.strip():
-            raise EmptyOrganizationNameError()
-        return name.strip()
-
-    @classmethod
-    async def as_form(cls, name: str = Form(...)):
-        return cls(name=name)
-
-
-class OrganizationRead(BaseModel):
-    model_config = ConfigDict(from_attributes=True)
-
-    id: int
-    name: str
-    created_at: datetime
-    updated_at: datetime
-
-
-class OrganizationUpdate(BaseModel):
-    id: int
-    name: str
-
-    @field_validator('name')
-    @classmethod
-    def validate_name(cls, name: str) -> str:
-        if not name.strip():
-            raise EmptyOrganizationNameError()
-        return name.strip()
-
-    @classmethod
-    async def as_form(cls, id: int = Form(...), name: str = Form(...)):
-        return cls(id=id, name=name)
-
-
 # --- Routes ---
 
 
@@ -84,26 +41,26 @@ async def read_organization(
 
 @router.post("/create", response_class=RedirectResponse)
 def create_organization(
-    org: OrganizationCreate = Depends(OrganizationCreate.as_form),
+    name: Annotated[str, StringConstraints(min_length=1, strip_whitespace=True)] = Form(...),
     user: User = Depends(get_authenticated_user),
     session: Session = Depends(get_session)
 ) -> RedirectResponse:
     # Check if organization already exists
     db_org = session.exec(select(Organization).where(
-        Organization.name == org.name)).first()
+        Organization.name == name)).first()
     if db_org:
         raise OrganizationNameTakenError()
 
     # Create organization first
-    db_org = Organization(name=org.name)
+    db_org = Organization(name=name)
     session.add(db_org)
     # This gets us the org ID without committing
     session.flush()
 
     # Create default roles with organization_id
     initial_roles = [
-        Role(name=name, organization_id=db_org.id)
-        for name in default_roles
+        Role(name=role_name, organization_id=db_org.id)
+        for role_name in default_roles
     ]
     session.add_all(initial_roles)
     session.flush()
@@ -124,7 +81,7 @@ def create_organization(
 @router.post("/update/{org_id}", name="update_organization", response_class=RedirectResponse)
 def update_organization(
     org_id: int,
-    org: OrganizationUpdate = Depends(OrganizationUpdate.as_form),
+    name: Annotated[str, StringConstraints(min_length=1, strip_whitespace=True)] = Form(...),
     user: User = Depends(get_user_with_relations),
     session: Session = Depends(get_session)
 ) -> RedirectResponse:
@@ -139,14 +96,14 @@ def update_organization(
     # Check if new name already exists for another organization
     existing_org = session.exec(
         select(Organization)
-        .where(Organization.name == org.name)
-        .where(Organization.id != org.id)
+        .where(Organization.name == name)
+        .where(Organization.id != org_id)
     ).first()
     if existing_org:
         raise OrganizationNameTakenError()
 
     # Update organization name
-    organization.name = org.name
+    organization.name = name
     organization.updated_at = utc_time()
     session.add(organization)
     session.commit()

routers/role.py

@@ -4,7 +4,6 @@
 from logging import getLogger
 from fastapi import APIRouter, Depends, Form
 from fastapi.responses import RedirectResponse
-from pydantic import BaseModel, ConfigDict
 from sqlmodel import Session, select, col
 from sqlalchemy.orm import selectinload
 from utils.db import get_session
@@ -16,102 +15,39 @@
 
 router = APIRouter(prefix="/roles", tags=["roles"])
 
-# --- Server Request Models ---
-
-class RoleCreate(BaseModel):
-    model_config = ConfigDict(from_attributes=True)
-
-    name: str
-    organization_id: int
-    permissions: List[ValidPermissions]
-
-    @classmethod
-    async def as_form(
-        cls,
-        name: str = Form(...),
-        organization_id: int = Form(...),
-        permissions: List[ValidPermissions] = Form(...)
-    ):
-        # Pass session to validator context
-        return cls(
-            name=name,
-            organization_id=organization_id,
-            permissions=permissions
-        )
-
-
-class RoleUpdate(BaseModel):
-    model_config = ConfigDict(from_attributes=True)
-
-    id: int
-    name: str
-    organization_id: int
-    permissions: List[ValidPermissions]
-
-    @classmethod
-    async def as_form(
-        cls,
-        id: int = Form(...),
-        name: str = Form(...),
-        organization_id: int = Form(...),
-        permissions: List[ValidPermissions] = Form(...)
-    ):
-        return cls(
-            id=id,
-            name=name,
-            organization_id=organization_id,
-            permissions=permissions
-        )
-
-
-class RoleDelete(BaseModel):
-    model_config = ConfigDict(from_attributes=True)
-
-    id: int
-    organization_id: int
-
-    @classmethod
-    async def as_form(
-        cls,
-        id: int = Form(...),
-        organization_id: int = Form(...)
-    ):
-        return cls(id=id, organization_id=organization_id)
-
-
 # --- Routes ---
 
-
 @router.post("/create", response_class=RedirectResponse)
 def create_role(
-    role: RoleCreate = Depends(RoleCreate.as_form),
+    name: str = Form(...),
+    organization_id: int = Form(...),
     user: User = Depends(get_authenticated_user),
     session: Session = Depends(get_session)
 ) -> RedirectResponse:
     # Check that the user-selected role name is unique for the organization
     if session.exec(
         select(Role).where(
-            Role.name == role.name,
-            Role.organization_id == role.organization_id
+            Role.name == name,
+            Role.organization_id == organization_id
         )
     ).first():
         raise RoleAlreadyExistsError()
 
     # Check that the user is authorized to create roles in the organization
-    if not user.has_permission(ValidPermissions.CREATE_ROLE, role.organization_id):
+    if not user.has_permission(ValidPermissions.CREATE_ROLE, organization_id):
         raise InsufficientPermissionsError()
 
     # Create role
     db_role = Role(
-        name=role.name,
-        organization_id=role.organization_id
+        name=name,
+        organization_id=organization_id
     )
     session.add(db_role)
 
     # Select Permission records corresponding to the user-selected permissions
     # and associate them with the newly created role
     permissions: Sequence[Permission] = session.exec(
-        select(Permission).where(col(Permission.name).in_(role.permissions))
+        select(Permission).where(col(Permission.name).in_(permissions))
     ).all()
     db_role.permissions.extend(permissions)
 
@@ -123,30 +59,33 @@ def create_role(
 
 @router.post("/update", response_class=RedirectResponse)
 def update_role(
-    role: RoleUpdate = Depends(RoleUpdate.as_form),
+    id: int = Form(...),
+    name: str = Form(...),
+    organization_id: int = Form(...),
+    permissions: List[ValidPermissions] = Form(...),
     user: User = Depends(get_authenticated_user),
     session: Session = Depends(get_session)
 ) -> RedirectResponse:
     # Check that the user is authorized to update the role
-    if not user.has_permission(ValidPermissions.EDIT_ROLE, role.organization_id):
+    if not user.has_permission(ValidPermissions.EDIT_ROLE, organization_id):
         raise InsufficientPermissionsError()
 
     # Select db_role to update, along with its permissions, by ID
     db_role: Optional[Role] = session.exec(
-        select(Role).where(Role.id == role.id).options(
+        select(Role).where(Role.id == id).options(
             selectinload(Role.permissions))
     ).first()
 
     if not db_role:
         raise RoleNotFoundError()
 
     # If any user-selected permissions are not valid, raise an error
-    for permission in role.permissions:
+    for permission in permissions:
         if permission not in ValidPermissions:
             raise InvalidPermissionError(permission)
 
     # Add any user-selected permissions that are not already associated with the role
-    for permission in role.permissions:
+    for permission in permissions:
         if permission not in [p.name for p in db_role.permissions]:
             db_permission: Optional[Permission] = session.exec(
                 select(Permission).where(Permission.name == permission)
@@ -158,21 +97,21 @@ def update_role(
 
     # Remove any permissions that are not user-selected
     for db_permission in db_role.permissions:
-        if db_permission.name not in role.permissions:
+        if db_permission.name not in permissions:
             db_role.permissions.remove(db_permission)
 
     # Check that no existing organization role has the same name but a different ID
     if session.exec(
         select(Role).where(
-            Role.name == role.name,
-            Role.organization_id == role.organization_id,
-            Role.id != role.id
+            Role.name == name,
+            Role.organization_id == organization_id,
+            Role.id != id
         )
     ).first():
         raise RoleAlreadyExistsError()
 
     # Update role name and updated_at timestamp
-    db_role.name = role.name
+    db_role.name = name
     db_role.updated_at = utc_time()
 
     session.commit()
@@ -182,17 +121,18 @@ def update_role(
 
 @router.post("/delete", response_class=RedirectResponse)
 def delete_role(
-    role: RoleDelete = Depends(RoleDelete.as_form),
+    id: int = Form(...),
+    organization_id: int = Form(...),
     user: User = Depends(get_authenticated_user),
     session: Session = Depends(get_session)
 ) -> RedirectResponse:
     # Check that the user is authorized to delete the role
-    if not user.has_permission(ValidPermissions.DELETE_ROLE, role.organization_id):
+    if not user.has_permission(ValidPermissions.DELETE_ROLE, organization_id):
         raise InsufficientPermissionsError()
 
     # Select the role to delete by ID, along with its users
     db_role: Role | None = session.exec(
-        select(Role).where(Role.id == role.id).options(
+        select(Role).where(Role.id == id).options(
             selectinload(Role.users)
         )
     ).first()