HR-ATS-Portal/backend/role/models.py

from datetime import datetime, timezone
from enum import Enum
from sqlalchemy import Column, DateTime, UniqueConstraint, func, or_
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.ext.asyncio import AsyncSession
from sqlmodel import Field, Relationship, SQLModel, select


def _now() -> datetime:
    return datetime.now(timezone.utc)


class EnumRoles(str, Enum):
    """Canonical keys for the eight seeded system roles. `Roles.role_name` is a varchar."""

    SYSTEM_ADMINISTRATOR = "system_administrator"
    HR_ADMINISTRATOR = "hr_administrator"
    RECRUITER = "recruiter"
    HIRING_MANAGER = "hiring_manager"
    DEPARTMENT_HEAD = "department_head"
    INTERVIEWER = "interviewer"
    CEO = "ceo"
    CANDIDATE = "candidate"


class PermissionTags(SQLModel, table=True):
    __tablename__ = "permission_tags"
    __table_args__ = (
        UniqueConstraint("module", "action", name="uq_permission_tags_module_action"),
    )

    id: int | None = Field(default=None, primary_key=True)
    tag_name: str = Field(max_length=64, unique=True, nullable=False, index=True)
    module: str = Field(max_length=32, nullable=False, index=True)
    action: str = Field(max_length=32, nullable=False)
    description: str | None = Field(default=None)
    created_at: datetime = Field(default_factory=_now, sa_type=DateTime(timezone=True))
    updated_at: datetime = Field(default_factory=_now, sa_type=DateTime(timezone=True))
    is_active: bool = Field(default=True)
    is_deleted: bool = Field(default=False)

    @classmethod
    def _search_filter(cls, search: str):
        pattern = f"%{search}%"
        return or_(
            cls.tag_name.ilike(pattern),
            cls.module.ilike(pattern),
            cls.action.ilike(pattern),
            cls.description.ilike(pattern),
        )

    @classmethod
    async def get_permission_tags(
        cls, session: AsyncSession, top: int | None, skip: int, search: str | None
    ):
        statement = (
            select(cls)
            .where(cls.is_deleted == False)  # noqa: E712
            .order_by(cls.module.asc(), cls.action.asc())
        )
        if search:
            statement = statement.where(cls._search_filter(search))
        if skip:
            statement = statement.offset(skip)
        if top is not None:
            statement = statement.limit(top)
        result = await session.execute(statement)
        return result.scalars().all()

    @classmethod
    async def get_permission_tag_by_id(cls, session: AsyncSession, record_id: int):
        statement = select(cls).where(cls.id == record_id)
        result = await session.execute(statement)
        return result.scalars().first()

    @classmethod
    async def get_permission_tags_by_ids(cls, session: AsyncSession, ids: list[int]):
        if not ids:
            return []
        statement = select(cls).where(
            cls.id.in_(ids),
            cls.is_active == True,  # noqa: E712
            cls.is_deleted == False,  # noqa: E712
        )
        result = await session.execute(statement)
        return result.scalars().all()

    @classmethod
    async def count_permission_tags(cls, session: AsyncSession, search: str | None):
        statement = (
            select(func.count())
            .select_from(cls)
            .where(cls.is_deleted == False)  # noqa: E712
        )
        if search:
            statement = statement.where(cls._search_filter(search))
        result = await session.execute(statement)
        return result.scalar_one()


class Permissions(SQLModel, table=True):
    """Named permission bundles — each row holds a JSONB array of permission_tags.id."""

    __tablename__ = "permissions"

    id: int | None = Field(default=None, primary_key=True)
    name: str = Field(max_length=64, unique=True, nullable=False)
    description: str | None = Field(default=None)
    permission_tags: list | None = Field(default=None, sa_column=Column(JSONB))
    is_system: bool = Field(default=False)
    created_at: datetime = Field(default_factory=_now, sa_type=DateTime(timezone=True))
    updated_at: datetime = Field(default_factory=_now, sa_type=DateTime(timezone=True))
    is_active: bool = Field(default=True)
    is_deleted: bool = Field(default=False)

    @classmethod
    def _search_filter(cls, search: str):
        pattern = f"%{search}%"
        return or_(cls.name.ilike(pattern), cls.description.ilike(pattern))

    @classmethod
    async def get_permissions(
        cls, session: AsyncSession, top: int | None, skip: int, search: str | None
    ):
        statement = (
            select(cls)
            .where(cls.is_deleted == False)  # noqa: E712
            .order_by(cls.name.asc())
        )
        if search:
            statement = statement.where(cls._search_filter(search))
        if skip:
            statement = statement.offset(skip)
        if top is not None:
            statement = statement.limit(top)
        result = await session.execute(statement)
        return result.scalars().all()

    @classmethod
    async def get_permission_by_id(cls, session: AsyncSession, record_id: int):
        statement = select(cls).where(cls.id == record_id)
        result = await session.execute(statement)
        return result.scalars().first()

    @classmethod
    async def get_permission_by_name(cls, session: AsyncSession, name: str):
        statement = select(cls).where(cls.name == name, cls.is_deleted == False)  # noqa: E712
        result = await session.execute(statement)
        return result.scalars().first()

    @classmethod
    async def get_permissions_by_ids(cls, session: AsyncSession, ids: list[int]):
        if not ids:
            return []
        statement = select(cls).where(
            cls.id.in_(ids),
            cls.is_active == True,  # noqa: E712
            cls.is_deleted == False,  # noqa: E712
        )
        result = await session.execute(statement)
        return result.scalars().all()

    @classmethod
    async def count_permissions(cls, session: AsyncSession, search: str | None):
        statement = (
            select(func.count())
            .select_from(cls)
            .where(cls.is_deleted == False)  # noqa: E712
        )
        if search:
            statement = statement.where(cls._search_filter(search))
        result = await session.execute(statement)
        return result.scalar_one()

    @classmethod
    async def insert_permission(cls, session: AsyncSession, fields: dict):
        row = cls(**fields)
        session.add(row)
        await session.commit()
        return await cls.get_permission_by_id(session, row.id)

    @classmethod
    async def update_permission(cls, session: AsyncSession, record_id: int, fields: dict):
        row = await cls.get_permission_by_id(session, record_id)
        if not row:
            return None
        for key, value in fields.items():
            setattr(row, key, value)
        row.updated_at = _now()
        session.add(row)
        await session.commit()
        await session.refresh(row)
        return row

    @classmethod
    async def soft_delete_permission(cls, session: AsyncSession, record_id: int):
        row = await cls.get_permission_by_id(session, record_id)
        if not row:
            return None
        row.is_deleted = True
        row.is_active = False
        row.updated_at = _now()
        session.add(row)
        await session.commit()
        await session.refresh(row)
        return row


class Roles(SQLModel, table=True):
    __tablename__ = "roles"

    id: int | None = Field(default=None, primary_key=True)
    role_name: str = Field(max_length=64, unique=True, nullable=False)
    description: str | None = Field(default=None)
    permissions: list | None = Field(default=None, sa_column=Column(JSONB))
    is_system: bool = Field(default=False)
    created_at: datetime = Field(default_factory=_now, sa_type=DateTime(timezone=True))
    updated_at: datetime = Field(default_factory=_now, sa_type=DateTime(timezone=True))
    is_active: bool = Field(default=True)
    is_deleted: bool = Field(default=False)

    users: list["Users"] = Relationship(back_populates="role")

    @classmethod
    def _search_filter(cls, search: str):
        pattern = f"%{search}%"
        return or_(cls.role_name.ilike(pattern), cls.description.ilike(pattern))

    @classmethod
    async def get_roles(
        cls, session: AsyncSession, top: int | None, skip: int, search: str | None
    ):
        statement = (
            select(cls)
            .where(cls.is_deleted == False)  # noqa: E712
            .order_by(cls.role_name.asc())
        )
        if search:
            statement = statement.where(cls._search_filter(search))
        if skip:
            statement = statement.offset(skip)
        if top is not None:
            statement = statement.limit(top)
        result = await session.execute(statement)
        return result.scalars().all()

    @classmethod
    async def get_role_by_id(cls, session: AsyncSession, record_id: int):
        statement = select(cls).where(cls.id == record_id)
        result = await session.execute(statement)
        return result.scalars().first()

    @classmethod
    async def get_role_by_name(cls, session: AsyncSession, role_name: str):
        statement = select(cls).where(
            cls.role_name == role_name, cls.is_deleted == False  # noqa: E712
        )
        result = await session.execute(statement)
        return result.scalars().first()

    @classmethod
    async def get_by_names(cls, session: AsyncSession, names):
        keys = [n for n in (names or []) if n]
        if not keys:
            return []
        result = await session.execute(
            select(cls).where(cls.role_name.in_(keys), cls.is_deleted == False)  # noqa: E712
        )
        return list(result.scalars().all())

    @classmethod
    async def count_roles(cls, session: AsyncSession, search: str | None):
        statement = (
            select(func.count())
            .select_from(cls)
            .where(cls.is_deleted == False)  # noqa: E712
        )
        if search:
            statement = statement.where(cls._search_filter(search))
        result = await session.execute(statement)
        return result.scalar_one()

    @classmethod
    async def insert_role(cls, session: AsyncSession, fields: dict):
        row = cls(**fields)
        session.add(row)
        await session.commit()
        return await cls.get_role_by_id(session, row.id)

    @classmethod
    async def update_role(cls, session: AsyncSession, record_id: int, fields: dict):
        row = await cls.get_role_by_id(session, record_id)
        if not row:
            return None
        for key, value in fields.items():
            setattr(row, key, value)
        row.updated_at = _now()
        session.add(row)
        await session.commit()
        await session.refresh(row)
        return row

    @classmethod
    async def soft_delete_role(cls, session: AsyncSession, record_id: int):
        row = await cls.get_role_by_id(session, record_id)
        if not row:
            return None
        row.is_deleted = True
        row.is_active = False
        row.updated_at = _now()
        session.add(row)
        await session.commit()
        await session.refresh(row)
        return row

    @classmethod
    async def resolve_tags(cls, session: AsyncSession, role: "Roles | None") -> tuple[str, ...]:
        """roles.permissions[] → permissions.permission_tags[] → permission_tags.tag_name.

        Dangling / inactive ids contribute nothing (deny, never error). NULL or [] denies all.
        """
        if role is None or not role.is_active or role.is_deleted:
            return ()
        perm_ids = role.permissions
        if not perm_ids or not isinstance(perm_ids, list):
            return ()
        bundles = await Permissions.get_permissions_by_ids(session, [int(i) for i in perm_ids])
        tag_ids: list[int] = []
        for bundle in bundles:
            raw = bundle.permission_tags
            if not raw or not isinstance(raw, list):
                continue
            tag_ids.extend(int(i) for i in raw)
        if not tag_ids:
            return ()
        tags = await PermissionTags.get_permission_tags_by_ids(session, tag_ids)
        # Stable unique order by tag id (seed order), then name as tiebreaker.
        ordered = sorted(tags, key=lambda t: (t.id or 0, t.tag_name))
        seen: set[str] = set()
        names: list[str] = []
        for tag in ordered:
            if tag.tag_name not in seen:
                seen.add(tag.tag_name)
                names.append(tag.tag_name)
        return tuple(names)


# Register Users so Roles.users Relationship can resolve (safe under circular import).
import users.models as _users_models  # noqa: E402, F401