Reorganize database schema to allow 3-way user-org-role relationship

Author: chriscarrollsmith

.gitignore

@@ -5,4 +5,5 @@ __pycache__
 /.quarto/
 _docs/
 .pytest_cache/
-.mypy_cache/
+.mypy_cache/
+.cursorrules

routers/organization.py

@@ -1,20 +1,66 @@
 from logging import getLogger
 from fastapi import APIRouter, Depends, HTTPException, Form
 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 Organization
+from utils.auth import get_authenticated_user
+from utils.models import Organization, User
 from datetime import datetime
 
 logger = getLogger("uvicorn.error")
 
+# -- Custom Exceptions --
+
+
+class EmptyOrganizationNameError(HTTPException):
+    def __init__(self):
+        super().__init__(
+            status_code=400,
+            detail="Organization name cannot be empty"
+        )
+
+
+class OrganizationExistsError(HTTPException):
+    def __init__(self):
+        super().__init__(
+            status_code=400,
+            detail="Organization already exists"
+        )
+
+
+class OrganizationNotFoundError(HTTPException):
+    def __init__(self):
+        super().__init__(
+            status_code=404,
+            detail="Organization not found"
+        )
+
+
+class OrganizationNameTakenError(HTTPException):
+    def __init__(self):
+        super().__init__(
+            status_code=400,
+            detail="Organization name already taken"
+        )
+
+
 router = APIRouter(prefix="/organizations", tags=["organizations"])
 
 
+# -- 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)
@@ -34,26 +80,30 @@ 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 --
+
 @router.post("/", response_class=RedirectResponse)
 def create_organization(
     org: OrganizationCreate = Depends(OrganizationCreate.as_form),
+    user: User = Depends(get_authenticated_user),
     session: Session = Depends(get_session)
 ) -> RedirectResponse:
-    # Validate organization name is not empty
-    if not org.name.strip():
-        raise HTTPException(
-            status_code=400, detail="Organization name cannot be empty")
-
     db_org = session.exec(select(Organization).where(
         Organization.name == org.name)).first()
     if db_org:
-        raise HTTPException(
-            status_code=400, detail="Organization already exists")
+        raise OrganizationExistsError()
 
     db_org = Organization(name=org.name)
     session.add(db_org)
@@ -64,26 +114,22 @@ def create_organization(
 
 
 @router.get("/{org_id}", response_model=OrganizationRead)
-def read_organization(org_id: int, session: Session = Depends(get_session)):
+def read_organization(org_id: int, user: User = Depends(get_authenticated_user), session: Session = Depends(get_session)):
     db_org = session.get(Organization, org_id)
     if not db_org:
-        raise HTTPException(status_code=404, detail="Organization not found")
+        raise OrganizationNotFoundError()
     return db_org
 
 
 @router.put("/{org_id}", response_class=RedirectResponse)
 def update_organization(
     org: OrganizationUpdate = Depends(OrganizationUpdate.as_form),
+    user: User = Depends(get_authenticated_user),
     session: Session = Depends(get_session)
 ) -> RedirectResponse:
-    # Validate organization name is not empty
-    if not org.name.strip():
-        raise HTTPException(
-            status_code=400, detail="Organization name cannot be empty")
-
     db_org = session.get(Organization, org.id)
     if not db_org:
-        raise HTTPException(status_code=404, detail="Organization not found")
+        raise OrganizationNotFoundError()
 
     # Check if new name already exists for another organization
     existing_org = session.exec(
@@ -92,8 +138,7 @@ def update_organization(
         .where(Organization.id != org.id)
     ).first()
     if existing_org:
-        raise HTTPException(
-            status_code=400, detail="Organization name already taken")
+        raise OrganizationNameTakenError()
 
     db_org.name = org.name
     db_org.updated_at = datetime.utcnow()
@@ -107,11 +152,12 @@ def update_organization(
 @router.delete("/{org_id}", response_class=RedirectResponse)
 def delete_organization(
     org_id: int,
+    user: User = Depends(get_authenticated_user),
     session: Session = Depends(get_session)
 ) -> RedirectResponse:
     db_org = session.get(Organization, org_id)
     if not db_org:
-        raise HTTPException(status_code=404, detail="Organization not found")
+        raise OrganizationNotFoundError()
 
     db_org.deleted = True
     db_org.updated_at = datetime.utcnow()

utils/models.py

@@ -24,28 +24,58 @@ class ValidPermissions(Enum):
     EDIT_ROLE = "Edit Role"
 
 
+class UserOrganizationLink(SQLModel, table=True):
+    id: Optional[int] = Field(default=None, primary_key=True)
+    user_id: int = Field(foreign_key="user.id")
+    organization_id: int = Field(foreign_key="organization.id")
+    role_id: int = Field(foreign_key="role.id")
+    created_at: datetime = Field(default_factory=utc_time)
+    updated_at: datetime = Field(default_factory=utc_time)
+
+    user: "User" = Relationship(back_populates="organization_links")
+    organization: "Organization" = Relationship(back_populates="user_links")
+    role: "Role" = Relationship(back_populates="user_links")
+
+
+class RolePermissionLink(SQLModel, table=True):
+    id: Optional[int] = Field(default=None, primary_key=True)
+    role_id: int = Field(foreign_key="role.id")
+    permission_id: int = Field(foreign_key="permission.id")
+    created_at: datetime = Field(default_factory=utc_time)
+    updated_at: datetime = Field(default_factory=utc_time)
+
+
 class Organization(SQLModel, table=True):
     id: Optional[int] = Field(default=None, primary_key=True)
     name: str
     created_at: datetime = Field(default_factory=utc_time)
     updated_at: datetime = Field(default_factory=utc_time)
     deleted: bool = Field(default=False)
 
-    users: List["User"] = Relationship(back_populates="organization")
+    user_links: List[UserOrganizationLink] = Relationship(
+        back_populates="organization")
+    users: List["User"] = Relationship(
+        back_populates="organizations",
+        link_model=UserOrganizationLink
+    )
+    roles: List["Role"] = Relationship(back_populates="organization")
 
 
 class Role(SQLModel, table=True):
     id: Optional[int] = Field(default=None, primary_key=True)
     name: str
-    organization_id: Optional[int] = Field(
-        default=None, foreign_key="organization.id")
+    organization_id: int = Field(foreign_key="organization.id")
     created_at: datetime = Field(default_factory=utc_time)
     updated_at: datetime = Field(default_factory=utc_time)
     deleted: bool = Field(default=False)
 
-    users: List["User"] = Relationship(back_populates="role")
-    role_permission_links: List["RolePermissionLink"] = Relationship(
+    organization: Organization = Relationship(back_populates="roles")
+    user_links: List[UserOrganizationLink] = Relationship(
         back_populates="role")
+    permissions: List["Permission"] = Relationship(
+        back_populates="roles",
+        link_model=RolePermissionLink
+    )
 
 
 class Permission(SQLModel, table=True):
@@ -56,21 +86,10 @@ class Permission(SQLModel, table=True):
     updated_at: datetime = Field(default_factory=utc_time)
     deleted: bool = Field(default=False)
 
-    role_permission_links: List["RolePermissionLink"] = Relationship(
-        back_populates="permission")
-
-
-class RolePermissionLink(SQLModel, table=True):
-    id: Optional[int] = Field(default=None, primary_key=True)
-    role_id: Optional[int] = Field(
-        default=None, foreign_key="role.id")
-    permission_id: Optional[int] = Field(
-        default=None, foreign_key="permission.id")
-
-    role: Optional["Role"] = Relationship(
-        back_populates="role_permission_links")
-    permission: Optional["Permission"] = Relationship(
-        back_populates="role_permission_links")
+    roles: List["Role"] = Relationship(
+        back_populates="permissions",
+        link_model=RolePermissionLink
+    )
 
 
 class PasswordResetToken(SQLModel, table=True):
@@ -92,23 +111,15 @@ class User(SQLModel, table=True):
     email: str = Field(index=True, unique=True)
     hashed_password: str
     avatar_url: Optional[str] = None
-    organization_id: Optional[int] = Field(
-        default=None, foreign_key="organization.id")
-    role_id: Optional[int] = Field(default=None, foreign_key="role.id")
     created_at: datetime = Field(default_factory=utc_time)
     updated_at: datetime = Field(default_factory=utc_time)
     deleted: bool = Field(default=False)
 
-    organization: Optional["Organization"] = Relationship(
-        back_populates="users")
-    role: Optional["Role"] = Relationship(back_populates="users")
+    organization_links: List[UserOrganizationLink] = Relationship(
+        back_populates="user")
+    organizations: List["Organization"] = Relationship(
+        back_populates="users",
+        link_model=UserOrganizationLink
+    )
     password_reset_tokens: List["PasswordResetToken"] = Relationship(
         back_populates="user")
-
-
-class UserOrganizationLink(SQLModel, table=True):
-    id: Optional[int] = Field(default=None, primary_key=True)
-    user_id: Optional[int] = Field(
-        default=None, foreign_key="user.id")
-    organization_id: Optional[int] = Field(
-        default=None, foreign_key="organization.id")