340 lines
12 KiB
Python
340 lines
12 KiB
Python
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 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
|