arm-core/api/src/repositories/users.py

"""Репозиторий пользователей: личность и эффективные права одним запросом (spec authn-authz, D-5, R-17).

`effective_permissions(user) = ∪ permission.code по всем активным ролям пользователя` (CLAUDE.md: права — данные).
Код роли в проверках не участвует: роли возвращаются только для показа (`GET /me`, список администратора).
Запросы администрирования (spec users-admin): списки пользователей/ролей/прав, назначения, подсчёт обладателей права.
Функции не делают `commit`: транзакцией запроса управляет `SessionDep`.
"""

import uuid
from collections.abc import Collection, Sequence
from dataclasses import dataclass
from typing import Any

from sqlalchemy import Select, and_, delete, func, or_, select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import aliased

from src.models.access import AppUser, Permission, Role, RolePermission, UserRole
from src.models.directories import RegionalOffice
from src.repositories.directories import LIKE_ESCAPE, escape_like

# Достаточно двух строк, чтобы заметить неоднозначность; больше пользователей читать незачем
_AMBIGUITY_PROBE_LIMIT = 2


class AmbiguousUserError(Exception):
    """По email найдено больше одного пользователя (email в БД уникален с учётом регистра, сравнение — без него)."""


@dataclass(frozen=True)
class UserWithAccess:
    """Пользователь, его филиал, коды активных ролей и эффективные права (коды).

    `roles` — отсортированный кортеж кодов активных ролей; `permissions` — объединение прав этих ролей без дубликатов.
    """

    user: AppUser
    regional_office: RegionalOffice
    roles: tuple[str, ...]
    permissions: frozenset[str]


async def load_principal_data(session: AsyncSession, email: str) -> UserWithAccess | None:
    """Пользователь по email (без учёта регистра) вместе с филиалом, ролями и правами — ОДИН SQL-запрос.

    Возвращает `None`, если пользователя нет; бросает :class:`AmbiguousUserError`, если подходит больше одного
    (fail-closed, R-17). Неактивный пользователь возвращается как есть (решение — за вызывающим). Роль с
    `is_active=false` не даёт прав и в `roles` не попадает.
    """
    matched = (
        select(AppUser)
        .where(func.lower(AppUser.email) == email.lower())
        .order_by(AppUser.id)
        .limit(_AMBIGUITY_PROBE_LIMIT)
        .subquery()
    )
    user = aliased(AppUser, matched)
    statement = (
        select(user, RegionalOffice, Role.code, Permission.code)
        .join(RegionalOffice, RegionalOffice.id == user.regional_office_id)
        .outerjoin(UserRole, UserRole.user_id == user.id)
        .outerjoin(Role, and_(Role.id == UserRole.role_id, Role.is_active.is_(True)))
        .outerjoin(RolePermission, RolePermission.role_id == Role.id)
        .outerjoin(Permission, Permission.id == RolePermission.permission_id)
    )
    rows = (await session.execute(statement)).all()
    if not rows:
        return None
    users = {row[0].id for row in rows}
    if len(users) > 1:
        raise AmbiguousUserError(email)
    roles = {row[2] for row in rows if row[2] is not None}
    permissions = frozenset(row[3] for row in rows if row[3] is not None)
    return UserWithAccess(
        user=rows[0][0],
        regional_office=rows[0][1],
        roles=tuple(sorted(roles)),
        permissions=permissions,
    )


# --- Администрирование (spec users-admin, D-8) ----------------------------------------------------------------------


def _filtered_users(stmt: Select[Any], q: str | None, is_active: bool | None) -> Select[Any]:
    if q is not None and q.strip():
        pattern = f"%{escape_like(q.strip())}%"
        stmt = stmt.where(
            or_(AppUser.email.ilike(pattern, escape=LIKE_ESCAPE), AppUser.full_name.ilike(pattern, escape=LIKE_ESCAPE))
        )
    if is_active is not None:
        stmt = stmt.where(AppUser.is_active.is_(is_active))
    return stmt


async def list_users(
    session: AsyncSession, *, q: str | None, is_active: bool | None, limit: int, offset: int
) -> tuple[list[tuple[AppUser, RegionalOffice]], int]:
    """Страница пользователей с филиалом (по ФИО, затем id) и общее число по фильтру."""
    page_stmt = (
        _filtered_users(select(AppUser, RegionalOffice), q, is_active)
        .join(RegionalOffice, RegionalOffice.id == AppUser.regional_office_id)
        .order_by(AppUser.full_name, AppUser.id)
        .limit(limit)
        .offset(offset)
    )
    total_stmt = _filtered_users(select(func.count(AppUser.id)), q, is_active)
    rows = (await session.execute(page_stmt)).all()
    total = (await session.execute(total_stmt)).scalar_one()
    return [(user, office) for user, office in rows], total


async def get_user(session: AsyncSession, user_id: uuid.UUID) -> AppUser | None:
    """Пользователь по id."""
    return await session.get(AppUser, user_id)


async def email_taken(session: AsyncSession, normalized_email: str) -> bool:
    """Есть ли пользователь с таким email без учёта регистра (`normalized_email` — уже `strip().lower()`)."""
    stmt = select(AppUser.id).where(func.lower(AppUser.email) == normalized_email).limit(1)
    return (await session.execute(stmt)).first() is not None


async def role_codes_by_user(session: AsyncSession, user_ids: Sequence[uuid.UUID]) -> dict[uuid.UUID, list[str]]:
    """Коды НАЗНАЧЕННЫХ ролей (включая неактивные роли) по пользователям, по алфавиту."""
    if not user_ids:
        return {}
    stmt = (
        select(UserRole.user_id, Role.code)
        .join(Role, Role.id == UserRole.role_id)
        .where(UserRole.user_id.in_(user_ids))
        .order_by(Role.code)
    )
    result: dict[uuid.UUID, list[str]] = {user_id: [] for user_id in user_ids}
    for user_id, code in (await session.execute(stmt)).all():
        result[user_id].append(code)
    return result


async def user_roles(session: AsyncSession, user_id: uuid.UUID) -> list[tuple[UserRole, Role]]:
    """Назначения ролей пользователя вместе с ролями."""
    stmt = select(UserRole, Role).join(Role, Role.id == UserRole.role_id).where(UserRole.user_id == user_id)
    return [(user_role, role) for user_role, role in (await session.execute(stmt)).all()]


async def delete_user_roles(session: AsyncSession, user_id: uuid.UUID, role_ids: Collection[uuid.UUID]) -> None:
    """Снимает с пользователя назначения указанных ролей."""
    if role_ids:
        await session.execute(delete(UserRole).where(UserRole.user_id == user_id, UserRole.role_id.in_(role_ids)))


async def roles_by_codes(session: AsyncSession, codes: Collection[str]) -> list[Role]:
    """Роли с указанными кодами (любого статуса)."""
    if not codes:
        return []
    return list((await session.execute(select(Role).where(Role.code.in_(codes)))).scalars().all())


async def get_role_by_code(session: AsyncSession, code: str) -> Role | None:
    """Роль по коду."""
    return (await session.execute(select(Role).where(Role.code == code))).scalar_one_or_none()


async def list_roles_with_permissions(session: AsyncSession) -> list[tuple[Role, list[str]]]:
    """Все роли (по коду) с кодами их прав (по алфавиту)."""
    roles = list((await session.execute(select(Role).order_by(Role.code))).scalars().all())
    stmt = (
        select(RolePermission.role_id, Permission.code)
        .join(Permission, Permission.id == RolePermission.permission_id)
        .order_by(Permission.code)
    )
    codes: dict[uuid.UUID, list[str]] = {role.id: [] for role in roles}
    for role_id, code in (await session.execute(stmt)).all():
        codes.setdefault(role_id, []).append(code)
    return [(role, codes[role.id]) for role in roles]


async def role_permissions(session: AsyncSession, role_id: uuid.UUID) -> list[Permission]:
    """Права роли (по коду)."""
    stmt = (
        select(Permission)
        .join(RolePermission, RolePermission.permission_id == Permission.id)
        .where(RolePermission.role_id == role_id)
        .order_by(Permission.code)
    )
    return list((await session.execute(stmt)).scalars().all())


async def delete_role_permissions(
    session: AsyncSession, role_id: uuid.UUID, permission_ids: Collection[uuid.UUID]
) -> None:
    """Снимает с роли указанные права."""
    if permission_ids:
        await session.execute(
            delete(RolePermission).where(
                RolePermission.role_id == role_id, RolePermission.permission_id.in_(permission_ids)
            )
        )


async def list_permissions(session: AsyncSession) -> list[Permission]:
    """Справочник прав (по коду)."""
    return list((await session.execute(select(Permission).order_by(Permission.code))).scalars().all())


async def permissions_by_codes(session: AsyncSession, codes: Collection[str]) -> list[Permission]:
    """Права с указанными кодами."""
    if not codes:
        return []
    return list((await session.execute(select(Permission).where(Permission.code.in_(codes)))).scalars().all())


async def lock_permission(session: AsyncSession, code: str) -> None:
    """`SELECT … FOR NO KEY UPDATE` строки права: сериализует проверки «последнего обладателя права» между транзакциями.

    Пока транзакция держит блокировку, конкурентная проверка того же права ждёт её завершения и затем считает уже
    зафиксированное состояние (READ COMMITTED: каждый оператор видит данные, зафиксированные до его начала).
    `NO KEY UPDATE` не мешает вставкам `role_permission` со ссылкой на это право (FK берёт `FOR KEY SHARE`).
    """
    await session.execute(select(Permission.id).where(Permission.code == code).with_for_update(key_share=True))


async def roles_grant_permission(session: AsyncSession, role_ids: Collection[uuid.UUID], code: str) -> bool:
    """Даёт ли хотя бы одна из АКТИВНЫХ ролей `role_ids` право `code`."""
    if not role_ids:
        return False
    stmt = (
        select(Role.id)
        .join(RolePermission, RolePermission.role_id == Role.id)
        .join(Permission, Permission.id == RolePermission.permission_id)
        .where(Role.id.in_(role_ids), Role.is_active.is_(True), Permission.code == code)
        .limit(1)
    )
    return (await session.execute(stmt)).first() is not None


async def count_active_users_with_permission(
    session: AsyncSession,
    code: str,
    *,
    excluding_user: uuid.UUID | None = None,
    excluding_role: uuid.UUID | None = None,
) -> int:
    """Число АКТИВНЫХ пользователей, которым АКТИВНЫЕ роли дают право `code` (как `effective_permissions`).

    `excluding_user` — не считать этого пользователя; `excluding_role` — не считать право, полученное через эту роль
    (пользователь учитывается, если право даёт другая его роль).
    """
    stmt = (
        select(func.count(func.distinct(AppUser.id)))
        .join(UserRole, UserRole.user_id == AppUser.id)
        .join(Role, and_(Role.id == UserRole.role_id, Role.is_active.is_(True)))
        .join(RolePermission, RolePermission.role_id == Role.id)
        .join(Permission, Permission.id == RolePermission.permission_id)
        .where(AppUser.is_active.is_(True), Permission.code == code)
    )
    if excluding_user is not None:
        stmt = stmt.where(AppUser.id != excluding_user)
    if excluding_role is not None:
        stmt = stmt.where(Role.id != excluding_role)
    return (await session.execute(stmt)).scalar_one()