Move private helper to shared utils and add custom exception message

Author: chriscarrollsmith

exceptions/http_exceptions.py

@@ -43,10 +43,11 @@ def __init__(self, field: str, message: str):
 
 
 class InsufficientPermissionsError(HTTPException):
-    def __init__(self):
-        super().__init__(
-            status_code=403, detail="You don't have permission to perform this action"
-        )
+    def __init__(
+        self,
+        message: str = "You don't have permission to perform this action",
+    ):
+        super().__init__(status_code=403, detail=message)
 
 
 class OrganizationSetupError(HTTPException):

routers/core/invitation.py

@@ -30,13 +30,9 @@
 )
 from exceptions.exceptions import EmailSendFailedError
 from utils.core.htmx import is_htmx_request, append_toast
-
-# Import the account router to generate URLs for login/register
+from utils.core.organizations import load_org_for_members_partial
 from routers.core.account import router as account_router
-from routers.core.organization import (
-    router as org_router,
-)  # Already imported, check usage
-from routers.core.user import _load_org_for_members_partial
+from routers.core.organization import router as org_router
 
 # Setup logger
 logger = getLogger("uvicorn.error")
@@ -85,7 +81,9 @@ async def create_invitation(
 
     # Check if the current user has permission to invite users to this organization
     if not current_user.has_permission(ValidPermissions.INVITE_USER, organization):
-        raise InsufficientPermissionsError()
+        raise InsufficientPermissionsError(
+            "You don't have permission to invite users to this organization"
+        )
 
     # Verify the role exists and belongs to this organization
     role = session.get(Role, role_id)
@@ -164,7 +162,7 @@ async def create_invitation(
     # HTMX: return partial; non-HTMX: PRG redirect
     if is_htmx_request(request):
         organization, user_permissions, active_invitations = (
-            _load_org_for_members_partial(session, organization_id, current_user)
+            load_org_for_members_partial(session, organization_id, current_user)
         )
         response = templates.TemplateResponse(
             request,
@@ -204,7 +202,9 @@ async def delete_invitation(
         raise OrganizationNotFoundError()
 
     if not current_user.has_permission(ValidPermissions.INVITE_USER, organization):
-        raise InsufficientPermissionsError()
+        raise InsufficientPermissionsError(
+            "You don't have permission to cancel invitations for this organization"
+        )
 
     invitation = session.get(Invitation, invitation_id)
     if not invitation or invitation.organization_id != organization_id:
@@ -215,7 +215,7 @@ async def delete_invitation(
 
     if is_htmx_request(request):
         organization, user_permissions, active_invitations = (
-            _load_org_for_members_partial(session, organization_id, current_user)
+            load_org_for_members_partial(session, organization_id, current_user)
         )
         response = templates.TemplateResponse(
             request,

routers/core/role.py

@@ -15,8 +15,8 @@
     utc_now,
     User,
     DataIntegrityError,
-    Organization,
 )
+from utils.core.organizations import load_org_for_roles_partial
 from utils.core.enums import ValidPermissions
 from utils.app.enums import AppPermissions
 from exceptions.http_exceptions import (
@@ -36,26 +36,6 @@
 templates = Jinja2Templates(directory="templates")
 
 
-def _load_org_for_roles_partial(
-    session: Session, organization_id: int, user: User
-) -> tuple:
-    """Re-query org with roles/users/permissions and compute user_permissions."""
-    organization = session.exec(
-        select(Organization)
-        .where(Organization.id == organization_id)
-        .options(
-            selectinload(Organization.roles).selectinload(Role.users),
-            selectinload(Organization.roles).selectinload(Role.permissions),
-        )
-    ).first()
-    user_permissions = set()
-    for role in user.roles:
-        if role.organization_id == organization_id:
-            for permission in role.permissions:
-                user_permissions.add(permission.name)
-    return organization, user_permissions
-
-
 # --- Routes ---
 
 
@@ -114,7 +94,7 @@ def create_role(
         raise RoleAlreadyExistsError()
 
     if is_htmx_request(request):
-        organization, user_permissions = _load_org_for_roles_partial(
+        organization, user_permissions = load_org_for_roles_partial(
             session, organization_id, user
         )
         response = templates.TemplateResponse(
@@ -221,7 +201,7 @@ def update_role(
     session.refresh(db_role)
 
     if is_htmx_request(request):
-        organization, user_permissions = _load_org_for_roles_partial(
+        organization, user_permissions = load_org_for_roles_partial(
             session, organization_id, user
         )
         response = templates.TemplateResponse(
@@ -282,7 +262,7 @@ def delete_role(
     session.commit()
 
     if is_htmx_request(request):
-        organization, user_permissions = _load_org_for_roles_partial(
+        organization, user_permissions = load_org_for_roles_partial(
             session, organization_id, user
         )
         response = templates.TemplateResponse(

routers/core/user.py

@@ -10,9 +10,8 @@
     AccountEmail,
     DataIntegrityError,
     Organization,
-    Role,
-    Invitation,
 )
+from utils.core.organizations import load_org_for_members_partial
 from utils.core.auth import MAX_EMAILS_PER_ACCOUNT
 from utils.core.dependencies import (
     get_authenticated_user,
@@ -40,32 +39,6 @@
 templates = Jinja2Templates(directory="templates")
 
 
-def _load_org_for_members_partial(
-    session: Session, organization_id: int, user: User
-) -> tuple:
-    """Re-query org with members fully loaded and compute user_permissions."""
-    organization = session.exec(
-        select(Organization)
-        .where(Organization.id == organization_id)
-        .options(
-            selectinload(Organization.roles)
-            .selectinload(Role.users)
-            .selectinload(User.account),
-            selectinload(Organization.roles)
-            .selectinload(Role.users)
-            .selectinload(User.roles),
-            selectinload(Organization.roles).selectinload(Role.permissions),
-        )
-    ).first()
-    user_permissions = set()
-    for role in user.roles:
-        if role.organization_id == organization_id:
-            for permission in role.permissions:
-                user_permissions.add(permission.name)
-    active_invitations = Invitation.get_active_for_org(session, organization_id)
-    return organization, user_permissions, active_invitations
-
-
 # --- Routes ---
 
 
@@ -267,7 +240,7 @@ def update_user_role(
 
     if is_htmx_request(request):
         organization, user_permissions, active_invitations = (
-            _load_org_for_members_partial(session, organization_id, user)
+            load_org_for_members_partial(session, organization_id, user)
         )
         response = templates.TemplateResponse(
             request,
@@ -341,7 +314,7 @@ def remove_user_from_organization(
 
     if is_htmx_request(request):
         organization, user_permissions, active_invitations = (
-            _load_org_for_members_partial(session, organization_id, user)
+            load_org_for_members_partial(session, organization_id, user)
         )
         response = templates.TemplateResponse(
             request,

utils/core/organizations.py

@@ -0,0 +1,51 @@
+from sqlmodel import Session, select
+from sqlalchemy.orm import selectinload
+
+from utils.core.models import Organization, Role, User, Invitation
+
+
+def _user_permissions_for_org(user: User, organization_id: int) -> set[str]:
+    user_permissions: set[str] = set()
+    for role in user.roles:
+        if role.organization_id == organization_id:
+            for permission in role.permissions:
+                user_permissions.add(permission.name)
+    return user_permissions
+
+
+def load_org_for_members_partial(
+    session: Session, organization_id: int, user: User
+) -> tuple[Organization | None, set[str], list[Invitation]]:
+    """Re-query org with members fully loaded and compute user_permissions."""
+    organization = session.exec(
+        select(Organization)
+        .where(Organization.id == organization_id)
+        .options(
+            selectinload(Organization.roles)
+            .selectinload(Role.users)
+            .selectinload(User.account),
+            selectinload(Organization.roles)
+            .selectinload(Role.users)
+            .selectinload(User.roles),
+            selectinload(Organization.roles).selectinload(Role.permissions),
+        )
+    ).first()
+    user_permissions = _user_permissions_for_org(user, organization_id)
+    active_invitations = Invitation.get_active_for_org(session, organization_id)
+    return organization, user_permissions, active_invitations
+
+
+def load_org_for_roles_partial(
+    session: Session, organization_id: int, user: User
+) -> tuple[Organization | None, set[str]]:
+    """Re-query org with roles/users/permissions and compute user_permissions."""
+    organization = session.exec(
+        select(Organization)
+        .where(Organization.id == organization_id)
+        .options(
+            selectinload(Organization.roles).selectinload(Role.users),
+            selectinload(Organization.roles).selectinload(Role.permissions),
+        )
+    ).first()
+    user_permissions = _user_permissions_for_org(user, organization_id)
+    return organization, user_permissions