From 4b623fc8cdc8c95a994db7905f1b63e22a434a7b Mon Sep 17 00:00:00 2001 From: as5584 Date: Thu, 20 Aug 2026 08:24:03 +0530 Subject: [PATCH] Add RBAC models and SQLAdmin views for issue #275 --- .../src/interfaces/admin/views/__init__.py | 10 +- .../admin/views/role_permissions.py | 38 ++++ backend/src/interfaces/admin/views/roles.py | 40 +++++ .../src/interfaces/admin/views/user_roles.py | 27 +++ backend/src/interfaces/admin/views/users.py | 10 +- backend/src/interfaces/api/v1/__init__.py | 3 +- backend/src/modules/__init__.py | 4 + backend/src/modules/common/exceptions.py | 5 + backend/src/modules/role/__init__.py | 12 ++ backend/src/modules/role/dependencies.py | 54 ++++++ backend/src/modules/role/models.py | 92 ++++++++++ backend/src/modules/role/permissions.py | 99 +++++++++++ backend/src/modules/role/routes.py | 165 ++++++++++++++++++ backend/src/modules/role/schemas.py | 109 ++++++++++++ backend/src/modules/role/service.py | 141 +++++++++++++++ backend/src/modules/user/models.py | 9 +- 16 files changed, 814 insertions(+), 4 deletions(-) create mode 100644 backend/src/interfaces/admin/views/role_permissions.py create mode 100644 backend/src/interfaces/admin/views/roles.py create mode 100644 backend/src/interfaces/admin/views/user_roles.py create mode 100644 backend/src/modules/role/__init__.py create mode 100644 backend/src/modules/role/dependencies.py create mode 100644 backend/src/modules/role/models.py create mode 100644 backend/src/modules/role/permissions.py create mode 100644 backend/src/modules/role/routes.py create mode 100644 backend/src/modules/role/schemas.py create mode 100644 backend/src/modules/role/service.py diff --git a/backend/src/interfaces/admin/views/__init__.py b/backend/src/interfaces/admin/views/__init__.py index 4995ea41..4c12a3a2 100644 --- a/backend/src/interfaces/admin/views/__init__.py +++ b/backend/src/interfaces/admin/views/__init__.py @@ -1,13 +1,18 @@ """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", ] @@ -15,4 +20,7 @@ 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) diff --git a/backend/src/interfaces/admin/views/role_permissions.py b/backend/src/interfaces/admin/views/role_permissions.py new file mode 100644 index 00000000..563c4a69 --- /dev/null +++ b/backend/src/interfaces/admin/views/role_permissions.py @@ -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 \ No newline at end of file diff --git a/backend/src/interfaces/admin/views/roles.py b/backend/src/interfaces/admin/views/roles.py new file mode 100644 index 00000000..c7058bf1 --- /dev/null +++ b/backend/src/interfaces/admin/views/roles.py @@ -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) \ No newline at end of file diff --git a/backend/src/interfaces/admin/views/user_roles.py b/backend/src/interfaces/admin/views/user_roles.py new file mode 100644 index 00000000..2e45da71 --- /dev/null +++ b/backend/src/interfaces/admin/views/user_roles.py @@ -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 \ No newline at end of file diff --git a/backend/src/interfaces/admin/views/users.py b/backend/src/interfaces/admin/views/users.py index d79214f6..078d1c12 100644 --- a/backend/src/interfaces/admin/views/users.py +++ b/backend/src/interfaces/admin/views/users.py @@ -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 @@ -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.""" @@ -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"] diff --git a/backend/src/interfaces/api/v1/__init__.py b/backend/src/interfaces/api/v1/__init__.py index 98f3b69e..85654f2c 100644 --- a/backend/src/interfaces/api/v1/__init__.py +++ b/backend/src/interfaces/api/v1/__init__.py @@ -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 @@ -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") diff --git a/backend/src/modules/__init__.py b/backend/src/modules/__init__.py index 186d38e1..739dff2c 100644 --- a/backend/src/modules/__init__.py +++ b/backend/src/modules/__init__.py @@ -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 @@ -12,4 +13,7 @@ "APIKey", "KeyUsage", "KeyPermission", + "Role", + "RolePermission", + "UserRole", ] diff --git a/backend/src/modules/common/exceptions.py b/backend/src/modules/common/exceptions.py index 3dc4d497..1875540d 100644 --- a/backend/src/modules/common/exceptions.py +++ b/backend/src/modules/common/exceptions.py @@ -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 diff --git a/backend/src/modules/role/__init__.py b/backend/src/modules/role/__init__.py new file mode 100644 index 00000000..74c00720 --- /dev/null +++ b/backend/src/modules/role/__init__.py @@ -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", +] \ No newline at end of file diff --git a/backend/src/modules/role/dependencies.py b/backend/src/modules/role/dependencies.py new file mode 100644 index 00000000..090edd69 --- /dev/null +++ b/backend/src/modules/role/dependencies.py @@ -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 \ No newline at end of file diff --git a/backend/src/modules/role/models.py b/backend/src/modules/role/models.py new file mode 100644 index 00000000..e1117b6a --- /dev/null +++ b/backend/src/modules/role/models.py @@ -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}" \ No newline at end of file diff --git a/backend/src/modules/role/permissions.py b/backend/src/modules/role/permissions.py new file mode 100644 index 00000000..b5eca1bf --- /dev/null +++ b/backend/src/modules/role/permissions.py @@ -0,0 +1,99 @@ +"""Permission constants and hierarchical permission tree. + +Permissions are code constants, not database rows. Roles store permission +names as strings on ``RolePermission.permission_name``. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field + + +class PermissionNames: + """Central list of permission name constants.""" + + user = "user" + user_read = f"{user}.read" + user_create = f"{user}.create" + user_update = f"{user}.update" + user_delete = f"{user}.delete" + + role = "role" + role_read = f"{role}.read" + role_create = f"{role}.create" + role_update = f"{role}.update" + role_delete = f"{role}.delete" + role_assign = f"{role}.assign" + + tier = "tier" + tier_read = f"{tier}.read" + tier_create = f"{tier}.create" + tier_update = f"{tier}.update" + tier_delete = f"{tier}.delete" + + +@dataclass(frozen=True) +class PermissionNode: + """One node in the permission tree (used by admin UI and docs).""" + + name: str + children: tuple[PermissionNode, ...] = field(default_factory=tuple) + + +PERMISSION_TREE: tuple[PermissionNode, ...] = ( + PermissionNode( + name=PermissionNames.user, + children=( + PermissionNode(name=PermissionNames.user_read), + PermissionNode(name=PermissionNames.user_create), + PermissionNode(name=PermissionNames.user_update), + PermissionNode(name=PermissionNames.user_delete), + ), + ), + PermissionNode( + name=PermissionNames.role, + children=( + PermissionNode(name=PermissionNames.role_read), + PermissionNode(name=PermissionNames.role_create), + PermissionNode(name=PermissionNames.role_update), + PermissionNode(name=PermissionNames.role_delete), + PermissionNode(name=PermissionNames.role_assign), + ), + ), + PermissionNode( + name=PermissionNames.tier, + children=( + PermissionNode(name=PermissionNames.tier_read), + PermissionNode(name=PermissionNames.tier_create), + PermissionNode(name=PermissionNames.tier_update), + PermissionNode(name=PermissionNames.tier_delete), + ), + ), +) + + +def flatten_permission_tree(tree: tuple[PermissionNode, ...] = PERMISSION_TREE) -> list[str]: + """Return permission names in depth-first order (parents before children).""" + + names: list[str] = [] + + def walk(node: PermissionNode) -> None: + names.append(node.name) + for child in node.children: + walk(child) + + for root in tree: + walk(root) + return names + + +def is_known_permission(name: str) -> bool: + """Return True if ``name`` is defined in the permission tree.""" + + return name in set(flatten_permission_tree()) + + +def permission_choices() -> list[tuple[str, str]]: + """WTForms choices for the admin permission dropdown.""" + + return [(name, name) for name in flatten_permission_tree()] \ No newline at end of file diff --git a/backend/src/modules/role/routes.py b/backend/src/modules/role/routes.py new file mode 100644 index 00000000..e77d9cba --- /dev/null +++ b/backend/src/modules/role/routes.py @@ -0,0 +1,165 @@ +from typing import Annotated, Any + +from fastapi import APIRouter, Depends, HTTPException +from fastcrud import PaginatedListResponse, compute_offset, paginated_response + +from ...infrastructure.auth.http_exceptions import ForbiddenException, NotFoundException +from ...infrastructure.dependencies import AsyncSessionDep +from ..common.exceptions import PermissionDeniedError, ResourceExistsError, RoleNotFoundError, ValidationError +from ..common.utils.error_handler import handle_exception +from .dependencies import RoleServiceDep, require_permissions +from .permissions import PermissionNames, flatten_permission_tree +from .schemas import RoleCreate, RolePermissionCreate, RoleRead, RoleUpdate, UserRoleCreate + +router = APIRouter(tags=["Roles"]) + +RequireRoleRead = Annotated[dict[str, Any], Depends(require_permissions(PermissionNames.role_read))] +RequireRoleCreate = Annotated[dict[str, Any], Depends(require_permissions(PermissionNames.role_create))] +RequireRoleUpdate = Annotated[dict[str, Any], Depends(require_permissions(PermissionNames.role_update))] +RequireRoleDelete = Annotated[dict[str, Any], Depends(require_permissions(PermissionNames.role_delete))] +RequireRoleAssign = Annotated[dict[str, Any], Depends(require_permissions(PermissionNames.role_assign))] + + +@router.get("/permissions", response_model=list[str], summary="List permission tree (flat)") +async def list_permissions() -> list[str]: + """Return every permission name from the code-defined tree.""" + return flatten_permission_tree() + + +@router.get("/", response_model=PaginatedListResponse[RoleRead], summary="List roles") +async def list_roles( + db: AsyncSessionDep, + role_service: RoleServiceDep, + _user: RequireRoleRead, + page: int = 1, + items_per_page: int = 10, +) -> dict: + try: + data = await role_service.get_all(db=db, skip=compute_offset(page, items_per_page), limit=items_per_page) + return paginated_response(crud_data=data, page=page, items_per_page=items_per_page) + except PermissionDeniedError as e: + raise ForbiddenException(str(e)) + except Exception as e: + http_exception = handle_exception(e) + if http_exception: + raise http_exception + raise HTTPException(status_code=500, detail="An unexpected error occurred") + + +@router.post("/", response_model=RoleRead, status_code=201, summary="Create a role") +async def create_role( + role: RoleCreate, + db: AsyncSessionDep, + role_service: RoleServiceDep, + _user: RequireRoleCreate, +) -> dict[str, Any]: + try: + return await role_service.create(role, db) + except ResourceExistsError as e: + raise HTTPException(status_code=409, detail=str(e)) + except PermissionDeniedError as e: + raise ForbiddenException(str(e)) + except Exception as e: + http_exception = handle_exception(e) + if http_exception: + raise http_exception + raise HTTPException(status_code=500, detail="An unexpected error occurred") + + +@router.get("/{role_id}", response_model=RoleRead, summary="Get a role") +async def get_role( + role_id: int, + db: AsyncSessionDep, + role_service: RoleServiceDep, + _user: RequireRoleRead, +) -> dict[str, Any]: + try: + return await role_service.get_by_id(role_id, db) + except RoleNotFoundError: + raise NotFoundException("Role not found") + except Exception as e: + http_exception = handle_exception(e) + if http_exception: + raise http_exception + raise HTTPException(status_code=500, detail="An unexpected error occurred") + + +@router.patch("/{role_id}", summary="Update a role") +async def update_role( + role_id: int, + role_update: RoleUpdate, + db: AsyncSessionDep, + role_service: RoleServiceDep, + _user: RequireRoleUpdate, +) -> dict[str, str]: + try: + await role_service.update(role_id, role_update, db) + return {"message": "Role updated"} + except RoleNotFoundError: + raise NotFoundException("Role not found") + except ResourceExistsError as e: + raise HTTPException(status_code=409, detail=str(e)) + except Exception as e: + http_exception = handle_exception(e) + if http_exception: + raise http_exception + raise HTTPException(status_code=500, detail="An unexpected error occurred") + + +@router.delete("/{role_id}", summary="Soft-delete a role") +async def delete_role( + role_id: int, + db: AsyncSessionDep, + role_service: RoleServiceDep, + _user: RequireRoleDelete, +) -> dict[str, str]: + try: + await role_service.soft_delete(role_id, db) + return {"message": "Role deleted"} + except RoleNotFoundError: + raise NotFoundException("Role not found") + except Exception as e: + http_exception = handle_exception(e) + if http_exception: + raise http_exception + raise HTTPException(status_code=500, detail="An unexpected error occurred") + + +@router.post("/{role_id}/permissions", status_code=201, summary="Assign a permission to a role") +async def assign_permission( + role_id: int, + body: RolePermissionCreate, + db: AsyncSessionDep, + role_service: RoleServiceDep, + _user: RequireRoleUpdate, +) -> dict[str, Any]: + payload = RolePermissionCreate(role_id=role_id, permission_name=body.permission_name) + try: + return await role_service.assign_permission(payload, db) + except (RoleNotFoundError, ValidationError, ResourceExistsError) as e: + status = 404 if isinstance(e, RoleNotFoundError) else 400 + raise HTTPException(status_code=status, detail=str(e)) + except Exception as e: + http_exception = handle_exception(e) + if http_exception: + raise http_exception + raise HTTPException(status_code=500, detail="An unexpected error occurred") + + +@router.post("/assignments", status_code=201, summary="Assign a role to a user") +async def assign_user_role( + body: UserRoleCreate, + db: AsyncSessionDep, + role_service: RoleServiceDep, + _user: RequireRoleAssign, +) -> dict[str, Any]: + try: + return await role_service.assign_role(body, db) + except (RoleNotFoundError, ResourceExistsError) as e: + status = 404 if isinstance(e, RoleNotFoundError) else 409 + raise HTTPException(status_code=status, detail=str(e)) + except Exception as e: + http_exception = handle_exception(e) + if http_exception: + raise http_exception + raise HTTPException(status_code=500, detail="An unexpected error occurred") \ No newline at end of file diff --git a/backend/src/modules/role/schemas.py b/backend/src/modules/role/schemas.py new file mode 100644 index 00000000..6b548e8e --- /dev/null +++ b/backend/src/modules/role/schemas.py @@ -0,0 +1,109 @@ +from datetime import datetime +from typing import Annotated + +from pydantic import BaseModel, Field + +from ..common.schemas import TimestampSchema + + +class RoleBase(BaseModel): + """Base role schema.""" + + name: Annotated[ + str, + Field( + description="Unique role name", + examples=["editor", "support"], + min_length=1, + max_length=50, + ), + ] + + +class Role(TimestampSchema, RoleBase): + """Complete role schema with timestamps.""" + + pass + + +class RoleSelect(BaseModel): + """Minimal schema for selecting required role fields.""" + + id: int + name: str + + +class RoleRead(RoleBase): + """Schema for reading a role.""" + + id: int + created_at: datetime + description: str | None = None + is_deleted: bool = False + permissions: list[str] = Field(default_factory=list) + + +class RoleCreate(RoleBase): + """Schema for creating a role.""" + + description: Annotated[ + str | None, + Field(description="Role description", max_length=255, default=None), + ] + + +class RoleCreateInternal(RoleCreate): + """Internal schema for role creation.""" + + pass + + +class RoleUpdate(BaseModel): + """Schema for updating a role.""" + + name: Annotated[ + str | None, + Field(description="Unique role name", min_length=1, max_length=50, default=None), + ] + description: Annotated[ + str | None, + Field(description="Role description", max_length=255, default=None), + ] + + +class RoleUpdateInternal(RoleUpdate): + """Internal schema for role updates.""" + + updated_at: datetime + + +class RolePermissionCreate(BaseModel): + """Assign a permission name to a role.""" + + role_id: int + permission_name: Annotated[str, Field(min_length=1, max_length=100)] + + +class RolePermissionRead(BaseModel): + """Read a role-permission assignment.""" + + id: int + role_id: int + permission_name: str + created_at: datetime + + +class UserRoleCreate(BaseModel): + """Assign a role to a user.""" + + user_id: int + role_id: int + + +class UserRoleRead(BaseModel): + """Read a user-role assignment.""" + + id: int + user_id: int + role_id: int + created_at: datetime \ No newline at end of file diff --git a/backend/src/modules/role/service.py b/backend/src/modules/role/service.py new file mode 100644 index 00000000..0702ba8d --- /dev/null +++ b/backend/src/modules/role/service.py @@ -0,0 +1,141 @@ +from typing import Any + +from fastcrud.types import GetMultiResponseDict +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from ..common.exceptions import PermissionDeniedError, ResourceExistsError, RoleNotFoundError, ValidationError +from .crud import crud_role_permissions, crud_roles, crud_user_roles +from .permissions import flatten_permission_tree, is_known_permission +from .schemas import ( + RoleCreate, + RoleCreateInternal, + RolePermissionCreate, + RoleRead, + RoleUpdate, + UserRoleCreate, +) + + +class RoleService: + """Business logic for roles, permission assignment, and user-role mapping.""" + + async def create(self, role: RoleCreate, db: AsyncSession) -> dict[str, Any]: + """Create a new role.""" + role_dict = role.model_dump() + if await crud_roles.exists(db=db, name=role_dict["name"]): + raise ResourceExistsError(f"Role with name '{role_dict['name']}' already exists") + + created = await crud_roles.create(db=db, object=RoleCreateInternal(**role_dict), schema_to_select=RoleRead) + if not created: + raise ResourceExistsError("Failed to create role") + return created + + async def get_all(self, db: AsyncSession, skip: int = 0, limit: int = 100) -> GetMultiResponseDict: + """Paginated list of non-deleted roles.""" + return await crud_roles.get_multi(db=db, offset=skip, limit=limit, schema_to_select=RoleRead, is_deleted=False) + + async def get_by_id(self, role_id: int, db: AsyncSession) -> dict[str, Any]: + """Retrieve a role by ID.""" + role = await crud_roles.get(db=db, id=role_id, schema_to_select=RoleRead, is_deleted=False) + if not role: + raise RoleNotFoundError(f"Role with ID {role_id} not found") + role["permissions"] = await self.list_permission_names(role_id, db) + return role + + async def get_by_name(self, name: str, db: AsyncSession) -> dict[str, Any]: + """Retrieve a role by name.""" + role = await crud_roles.get(db=db, name=name, schema_to_select=RoleRead, is_deleted=False) + if not role: + raise RoleNotFoundError(f"Role with name '{name}' not found") + role["permissions"] = await self.list_permission_names(role["id"], db) + return role + + async def update(self, role_id: int, role_update: RoleUpdate, db: AsyncSession) -> None: + """Update a role by ID.""" + existing = await crud_roles.get(db=db, id=role_id, schema_to_select=RoleRead) + if not existing: + raise RoleNotFoundError(f"Role with ID {role_id} not found") + + update_data = role_update.model_dump(exclude_unset=True) + if "name" in update_data and update_data["name"] != existing["name"]: + if await crud_roles.exists(db=db, name=update_data["name"]): + raise ResourceExistsError(f"Role with name '{update_data['name']}' already exists") + + await crud_roles.update(db=db, object=role_update, id=role_id) + + async def soft_delete(self, role_id: int, db: AsyncSession) -> None: + """Soft-delete a role.""" + existing = await crud_roles.get(db=db, id=role_id, schema_to_select=RoleRead, is_deleted=False) + if not existing: + raise RoleNotFoundError(f"Role with ID {role_id} not found") + await crud_roles.delete(db=db, id=role_id) + + async def list_permission_names(self, role_id: int, db: AsyncSession) -> list[str]: + """Return permission names assigned to a role.""" + rows = await crud_role_permissions.get_multi(db=db, role_id=role_id, limit=1000) + data = rows.get("data", []) if isinstance(rows, dict) else rows + return [row["permission_name"] for row in data] + + async def assign_permission(self, payload: RolePermissionCreate, db: AsyncSession) -> dict[str, Any]: + """Attach a known permission name to a role.""" + await self.get_by_id(payload.role_id, db) + if not is_known_permission(payload.permission_name): + raise ValidationError(f"Unknown permission '{payload.permission_name}'") + if await crud_role_permissions.exists(db=db, role_id=payload.role_id, permission_name=payload.permission_name): + raise ResourceExistsError("Permission already assigned to this role") + created = await crud_role_permissions.create(db=db, object=payload) + if not created: + raise ResourceExistsError("Failed to assign permission") + return created + + async def remove_permission(self, role_id: int, permission_name: str, db: AsyncSession) -> None: + """Remove a permission from a role.""" + existing = await crud_role_permissions.get(db=db, role_id=role_id, permission_name=permission_name) + if not existing: + raise RoleNotFoundError("Role permission assignment not found") + await crud_role_permissions.db_delete(db=db, id=existing["id"]) + + async def assign_role(self, payload: UserRoleCreate, db: AsyncSession) -> dict[str, Any]: + """Assign a role to a user.""" + await self.get_by_id(payload.role_id, db) + if await crud_user_roles.exists(db=db, user_id=payload.user_id, role_id=payload.role_id): + raise ResourceExistsError("User already has this role") + created = await crud_user_roles.create(db=db, object=payload) + if not created: + raise ResourceExistsError("Failed to assign role") + return created + + async def remove_role(self, user_id: int, role_id: int, db: AsyncSession) -> None: + """Remove a role from a user.""" + existing = await crud_user_roles.get(db=db, user_id=user_id, role_id=role_id) + if not existing: + raise RoleNotFoundError("User role assignment not found") + await crud_user_roles.db_delete(db=db, id=existing["id"]) + + async def get_user_role_ids(self, user_id: int, db: AsyncSession) -> list[int]: + """Return role IDs assigned to a user.""" + rows = await crud_user_roles.get_multi(db=db, user_id=user_id, limit=1000) + data = rows.get("data", []) if isinstance(rows, dict) else rows + return [row["role_id"] for row in data] + + async def get_effective_permissions(self, user_id: int, db: AsyncSession) -> set[str]: + """Union of permission names from all of a user's roles.""" + from .models import RolePermission, UserRole + + stmt = ( + select(RolePermission.permission_name) + .join(UserRole, UserRole.role_id == RolePermission.role_id) + .where(UserRole.user_id == user_id) + ) + result = await db.execute(stmt) + return set(result.scalars().all()) + + def all_permission_names(self) -> list[str]: + """All known permission names from the tree.""" + return flatten_permission_tree() + + def verify_superuser(self, user: dict[str, Any], action: str = "manage roles") -> None: + """Require superuser for privileged role management.""" + if not user.get("is_superuser", False): + raise PermissionDeniedError(f"Only superusers can {action}") \ No newline at end of file diff --git a/backend/src/modules/user/models.py b/backend/src/modules/user/models.py index 8c0d7eec..195b8468 100644 --- a/backend/src/modules/user/models.py +++ b/backend/src/modules/user/models.py @@ -9,7 +9,7 @@ if TYPE_CHECKING: from ..tier.models import Tier - + from ..role.models import UserRole class User(Base, TimestampMixin, SoftDeleteMixin): """User model representing application users.""" @@ -38,6 +38,13 @@ class User(Base, TimestampMixin, SoftDeleteMixin): index=True, default=None, ) + user_roles: Mapped[list["UserRole"]] = relationship( + "UserRole", + back_populates="user", + lazy="selectin", + default_factory=list, + init=False, + ) is_superuser: Mapped[bool] = mapped_column(default=False)