Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 9 additions & 1 deletion backend/src/interfaces/admin/views/__init__.py
Original file line number Diff line number Diff line change
@@ -1,18 +1,26 @@
"""SQLAdmin model views for the admin interface."""

from sqladmin import Admin

from .role_permissions import RolePermissionAdmin
from .roles import RoleAdmin
from .user_roles import UserRoleAdmin
from .tiers import TierAdmin
from .users import UserAdmin

__all__ = [
"UserAdmin",
"TierAdmin",
"RoleAdmin",
"RolePermissionAdmin",
"UserRoleAdmin",
"register_admin_views",
]


def register_admin_views(admin: Admin) -> None:
"""Register all model views with the admin interface."""
admin.add_view(UserAdmin)
admin.add_view(RoleAdmin)
admin.add_view(RolePermissionAdmin)
admin.add_view(UserRoleAdmin)
admin.add_view(TierAdmin)
38 changes: 38 additions & 0 deletions backend/src/interfaces/admin/views/role_permissions.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,38 @@
"""Admin view for role ↔ permission assignments."""

from sqladmin import ModelView
from wtforms import SelectField

from ....modules.role.models import RolePermission
from ....modules.role.permissions import permission_choices
from ..mixins import DataclassModelMixin


class RolePermissionAdmin(DataclassModelMixin, ModelView, model=RolePermission):
"""Assign permission-tree names to roles."""

name = "Role Permission"
name_plural = "Role Permissions"
icon = "fa-solid fa-key"
category = "Users & Access"

column_list = [
RolePermission.id,
RolePermission.role,
RolePermission.permission_name,
RolePermission.created_at,
]
column_labels = {"role": "Role", "permission_name": "Permission"}
column_searchable_list = [RolePermission.permission_name]
column_sortable_list = [RolePermission.id, RolePermission.permission_name]
column_default_sort = [(RolePermission.id, True)]

form_columns = [RolePermission.role_id, RolePermission.permission_name]
form_overrides = {"permission_name": SelectField}
form_args = {"permission_name": {"choices": permission_choices()}}

can_create = True
can_edit = True
can_delete = True
can_view_details = True
can_export = True
40 changes: 40 additions & 0 deletions backend/src/interfaces/admin/views/roles.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,40 @@
"""Admin view for Role model."""

from sqladmin import ModelView
from starlette.requests import Request

from ....infrastructure.database.session import local_session
from ....modules.role.models import Role
from ....modules.role.schemas import RoleCreate, RoleUpdate
from ....modules.role.service import RoleService
from ..mixins import DataclassModelMixin


class RoleAdmin(DataclassModelMixin, ModelView, model=Role):
"""Admin view for Role model."""

name = "Role"
name_plural = "Roles"
icon = "fa-solid fa-user-shield"
category = "Users & Access"

column_list = [Role.id, Role.name, Role.description, Role.created_at, Role.is_deleted]
column_details_list = "__all__"
column_searchable_list = [Role.name, Role.description]
column_sortable_list = [Role.id, Role.name, Role.created_at]
column_default_sort = [(Role.created_at, True)]
column_labels = {"is_deleted": "Deleted"}

can_create = True
can_edit = True
can_delete = True
can_view_details = True
can_export = True

form_create_rules = list(RoleCreate.model_fields.keys())
form_edit_rules = list(RoleUpdate.model_fields.keys())

async def delete_model(self, request: Request, pk: str) -> None:
"""Soft-delete the role through the service layer."""
async with local_session() as db:
await RoleService().soft_delete(role_id=int(pk), db=db)
27 changes: 27 additions & 0 deletions backend/src/interfaces/admin/views/user_roles.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,27 @@
"""Admin view for user ↔ role assignments."""

from sqladmin import ModelView

from ....modules.role.models import UserRole
from ..mixins import DataclassModelMixin


class UserRoleAdmin(DataclassModelMixin, ModelView, model=UserRole):
"""Assign roles to users."""

name = "User Role"
name_plural = "User Roles"
icon = "fa-solid fa-link"
category = "Users & Access"

column_list = [UserRole.id, UserRole.user, UserRole.role, UserRole.created_at]
column_labels = {"user": "User", "role": "Role"}
column_sortable_list = [UserRole.created_at]
column_default_sort = [(UserRole.created_at, True)]
form_columns = [UserRole.user_id, UserRole.role_id]

can_create = True
can_edit = True
can_delete = True
can_view_details = True
can_export = True
10 changes: 9 additions & 1 deletion backend/src/interfaces/admin/views/users.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,7 @@
from sqladmin import ModelView
from starlette.requests import Request
from wtforms import SelectField

from ....modules.role.dependencies import user_effective_permissions
from ....infrastructure.database.session import local_session
from ....modules.user.enums import OAuthProvider
from ....modules.user.models import User
Expand All @@ -16,6 +16,13 @@

OAUTH_PROVIDER_CHOICES = [("", "None")] + [(p.value, p.value.title()) for p in OAuthProvider]

def _format_assigned_roles(model, _attr) -> str:
roles = [
ur.role.name
for ur in getattr(model, "user_roles", []) or []
if getattr(ur, "role", None) is not None
]
return ", ".join(roles) if roles else "—"

class UserAdmin(DataclassModelMixin, ModelView, model=User):
"""Admin view for User model with password hashing."""
Expand All @@ -38,6 +45,7 @@ class UserAdmin(DataclassModelMixin, ModelView, model=User):
can_export = True

column_labels = {"hashed_password": "Password"}
column_formatters_detail = {"user_roles": _format_assigned_roles,}

form_create_rules = ["name", "username", "email", "hashed_password", "tier_id", "is_superuser"]
form_edit_rules = [*UserUpdate.model_fields.keys(), "tier_id", "is_superuser"]
Expand Down
3 changes: 2 additions & 1 deletion backend/src/interfaces/api/v1/__init__.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
from fastapi import APIRouter

from ....modules.role.routes import router as roles_router
from ....infrastructure.auth.routes import router as auth_router
from ....modules.api_keys.routes import router as api_keys_router
from ....modules.rate_limit.routes import router as rate_limits_router
Expand All @@ -12,3 +12,4 @@
router.include_router(rate_limits_router, prefix="/rate-limits")
router.include_router(auth_router, prefix="/auth")
router.include_router(api_keys_router, prefix="/api-keys")
router.include_router(roles_router, prefix="/roles")
4 changes: 4 additions & 0 deletions backend/src/modules/__init__.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
"""Initialize all modules and models to ensure SQLAlchemy registration."""

from .api_keys.models import APIKey, KeyPermission, KeyUsage
from .role.models import Role, RolePermission, UserRole
from .rate_limit.models import RateLimit
from .tier.models import Tier
from .user.models import User
Expand All @@ -12,4 +13,7 @@
"APIKey",
"KeyUsage",
"KeyPermission",
"Role",
"RolePermission",
"UserRole",
]
5 changes: 5 additions & 0 deletions backend/src/modules/common/exceptions.py
Original file line number Diff line number Diff line change
Expand Up @@ -65,3 +65,8 @@ class UsageLimitExceededError(DomainError):
"""Raised when a user exceeds their usage limits."""

pass

class RoleNotFoundError(ResourceNotFoundError):
"""Raised when a role or role assignment cannot be found."""

pass
12 changes: 12 additions & 0 deletions backend/src/modules/role/__init__.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,12 @@
from .models import Role, RolePermission, UserRole
from .permissions import PERMISSION_TREE, PermissionNames, PermissionNode, flatten_permission_tree

__all__ = [
"Role",
"RolePermission",
"UserRole",
"PermissionNames",
"PermissionNode",
"PERMISSION_TREE",
"flatten_permission_tree",
]
54 changes: 54 additions & 0 deletions backend/src/modules/role/dependencies.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,54 @@
from collections.abc import Callable
from typing import Annotated, Any

from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession

from ...infrastructure.auth.dependencies import get_current_user
from ...infrastructure.database.session import async_session
from ..common.exceptions import PermissionDeniedError
from .permissions import flatten_permission_tree
from .service import RoleService


def get_role_service() -> RoleService:
return RoleService()


RoleServiceDep = Annotated[RoleService, Depends(get_role_service)]


def require_permissions(*permission_names: str, require_all: bool = True) -> Callable[..., Any]:
"""FastAPI dependency factory that checks the caller's effective permissions."""

async def _dependency(
current_user: Annotated[dict[str, Any], Depends(get_current_user)],
db: Annotated[AsyncSession, Depends(async_session)],
role_service: RoleServiceDep,
) -> dict[str, Any]:
if current_user.get("is_superuser", False):
return current_user

granted = await role_service.get_effective_permissions(current_user["id"], db)
needed = set(permission_names)
ok = needed.issubset(granted) if require_all else bool(needed & granted)
if not ok:
raise PermissionDeniedError("Missing required permission(s)")
return current_user

return _dependency


def user_effective_permissions(user: Any) -> set[str]:
"""Flatten permissions from a loaded User ORM instance (admin display)."""
if getattr(user, "is_superuser", False):
return set(flatten_permission_tree())

names: set[str] = set()
for user_role in getattr(user, "user_roles", []) or []:
role = getattr(user_role, "role", None)
if role is None:
continue
for assignment in getattr(role, "permissions", []) or []:
names.add(assignment.permission_name)
return names
92 changes: 92 additions & 0 deletions backend/src/modules/role/models.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,92 @@
from typing import TYPE_CHECKING

from sqlalchemy import ForeignKey, Integer, String, UniqueConstraint
from sqlalchemy.orm import Mapped, mapped_column, relationship

from ...infrastructure.database.models import SoftDeleteMixin, TimestampMixin
from ...infrastructure.database.session import Base

if TYPE_CHECKING:
from ..user.models import User


class Role(Base, TimestampMixin, SoftDeleteMixin):
"""Reusable named role that holds a set of permission strings."""

__tablename__ = "roles"

id: Mapped[int] = mapped_column(
"id",
autoincrement=True,
nullable=False,
unique=True,
primary_key=True,
init=False,
)
name: Mapped[str] = mapped_column(String(50), nullable=False, unique=True, index=True)
description: Mapped[str | None] = mapped_column(String(255), default=None)

permissions: Mapped[list["RolePermission"]] = relationship(
"RolePermission",
back_populates="role",
lazy="selectin",
default_factory=list,
init=False,
)
user_roles: Mapped[list["UserRole"]] = relationship(
"UserRole",
back_populates="role",
lazy="selectin",
default_factory=list,
init=False,
)

def __repr__(self) -> str:
return self.name


class RolePermission(Base, TimestampMixin):
"""Maps a role to a permission name constant."""

__tablename__ = "role_permission"
__table_args__ = (UniqueConstraint("role_id", "permission_name", name="uq_role_permission"),)

id: Mapped[int] = mapped_column(
"id",
autoincrement=True,
nullable=False,
unique=True,
primary_key=True,
init=False,
)
role_id: Mapped[int] = mapped_column(Integer, ForeignKey("roles.id"), index=True)
permission_name: Mapped[str] = mapped_column(String(100), index=True)

role: Mapped["Role"] = relationship("Role", back_populates="permissions", lazy="selectin", init=False)

def __repr__(self) -> str:
return f"{self.role_id}:{self.permission_name}"


class UserRole(Base, TimestampMixin):
"""Maps a user to a role."""

__tablename__ = "user_role"
__table_args__ = (UniqueConstraint("user_id", "role_id", name="uq_user_role"),)

id: Mapped[int] = mapped_column(
"id",
autoincrement=True,
nullable=False,
unique=True,
primary_key=True,
init=False,
)
user_id: Mapped[int] = mapped_column(Integer, ForeignKey("user.id"), index=True)
role_id: Mapped[int] = mapped_column(Integer, ForeignKey("roles.id"), index=True)

user: Mapped["User"] = relationship("User", back_populates="user_roles", lazy="selectin", init=False)
role: Mapped["Role"] = relationship("Role", back_populates="user_roles", lazy="selectin", init=False)

def __repr__(self) -> str:
return f"user={self.user_id} role={self.role_id}"
Loading