124 lines
3.5 KiB
Python
124 lines
3.5 KiB
Python
"""Role management service."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
from datetime import UTC, datetime
|
|
from typing import Any
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.role import Role
|
|
|
|
|
|
class RoleService:
|
|
"""Handles role CRUD operations."""
|
|
|
|
async def list_roles(
|
|
self,
|
|
db: AsyncSession,
|
|
tenant_id: uuid.UUID,
|
|
) -> list[dict[str, Any]]:
|
|
"""List all roles in a tenant."""
|
|
q = select(Role).where(Role.tenant_id == tenant_id, Role.deleted_at.is_(None))
|
|
result = await db.execute(q)
|
|
roles = result.scalars().all()
|
|
return [self._role_to_dict(r) for r in roles]
|
|
|
|
async def create_role(
|
|
self,
|
|
db: AsyncSession,
|
|
tenant_id: uuid.UUID,
|
|
name: str,
|
|
permissions: dict[str, Any],
|
|
denied_permissions: list[str] | None = None,
|
|
field_permissions: dict[str, Any] | None = None,
|
|
) -> Role:
|
|
"""Create a new custom role."""
|
|
role = Role(
|
|
tenant_id=tenant_id,
|
|
name=name,
|
|
permissions=permissions,
|
|
denied_permissions=denied_permissions or [],
|
|
field_permissions=field_permissions or {},
|
|
)
|
|
db.add(role)
|
|
await db.flush()
|
|
return role
|
|
|
|
async def update_role(
|
|
self,
|
|
db: AsyncSession,
|
|
tenant_id: uuid.UUID,
|
|
role_id: uuid.UUID,
|
|
name: str | None = None,
|
|
permissions: dict[str, Any] | None = None,
|
|
denied_permissions: list[str] | None = None,
|
|
field_permissions: dict[str, Any] | None = None,
|
|
) -> Role | None:
|
|
"""Update a role. Increments permission_version on permission changes."""
|
|
q = select(Role).where(
|
|
Role.id == role_id,
|
|
Role.tenant_id == tenant_id,
|
|
Role.deleted_at.is_(None),
|
|
)
|
|
result = await db.execute(q)
|
|
role = result.scalar_one_or_none()
|
|
if role is None:
|
|
return None
|
|
|
|
version_bump = False
|
|
if name is not None:
|
|
role.name = name
|
|
if permissions is not None:
|
|
role.permissions = permissions
|
|
version_bump = True
|
|
if denied_permissions is not None:
|
|
role.denied_permissions = denied_permissions
|
|
version_bump = True
|
|
if field_permissions is not None:
|
|
role.field_permissions = field_permissions
|
|
version_bump = True
|
|
|
|
if version_bump:
|
|
role.permission_version += 1
|
|
|
|
await db.flush()
|
|
return role
|
|
|
|
async def delete_role(
|
|
self,
|
|
db: AsyncSession,
|
|
tenant_id: uuid.UUID,
|
|
role_id: uuid.UUID,
|
|
) -> bool:
|
|
"""Soft-delete a role."""
|
|
q = select(Role).where(
|
|
Role.id == role_id,
|
|
Role.tenant_id == tenant_id,
|
|
Role.deleted_at.is_(None),
|
|
)
|
|
result = await db.execute(q)
|
|
role = result.scalar_one_or_none()
|
|
if role is None:
|
|
return False
|
|
|
|
role.deleted_at = datetime.now(UTC)
|
|
await db.flush()
|
|
return True
|
|
|
|
def _role_to_dict(self, role: Role) -> dict[str, Any]:
|
|
"""Convert role to response dict."""
|
|
return {
|
|
"id": str(role.id),
|
|
"name": role.name,
|
|
"permissions": role.permissions,
|
|
"denied_permissions": role.denied_permissions,
|
|
"field_permissions": role.field_permissions,
|
|
"permission_version": role.permission_version,
|
|
}
|
|
|
|
|
|
role_service = RoleService()
|