leocrm/app/services/role_service.py

"""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()