api/modules/auth/api.py - master - systemsstrategyy/Treker

api/modules/auth/api.py
100 строк4 KB
SystemsStrategy
Initial import
02 июл 2026, 15:15
02 июл 2026, 15:159624ed1
"""Публичный интерфейс auth-модуля для остальных доменов.

Другие модули (Kanban, Comms, Billing, Quality) импортируют ТОЛЬКО отсюда.
Прямые импорты из `api.modules.auth.models` или `service` запрещены —
это нарушение границ модулей (см. конвенций разработки правило #4).

Здесь нет HTTP-эндпоинтов — это in-process API между модулями.
"""

from __future__ import annotations

from uuid import UUID

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from api.modules.auth.models import Organization, User
from api.shared.schemas import OrganizationPublic, UserDetail, UserPublic


class UserDirectory:
    """Чтение публичных данных юзеров. Используется Kanban для assignee,
    Comms для @mention'ов, и т.д."""

    @staticmethod
    async def get_user_public(db: AsyncSession, uid: UUID) -> UserPublic | None:
        """Возвращает публичный профиль по UID. None если юзер не найден."""
        user = await db.scalar(select(User).where(User.uid == uid))
        if user is None:
            return None
        await db.refresh(user, attribute_names=["organization"])
        return UserPublic(
            uid=user.uid,
            email=user.email,
            full_name=user.full_name,
            organization_uid=user.organization.uid,
        )

    @staticmethod
    async def list_users_in_org(db: AsyncSession, organization_uid: UUID) -> list[UserPublic]:
        """Список юзеров организации. Для дропдаунов assignee, mention'ов."""
        org = await db.scalar(select(Organization).where(Organization.uid == organization_uid))
        if org is None:
            return []

        users = (
            await db.scalars(
                select(User)
                .where(User.organization_id == org.id, User.is_active.is_(True))
                .order_by(User.full_name)
            )
        ).all()

        return [
            UserPublic(
                uid=u.uid,
                email=u.email,
                full_name=u.full_name,
                organization_uid=org.uid,
            )
            for u in users
        ]

    @staticmethod
    async def list_detailed_in_org(db: AsyncSession, organization_uid: UUID) -> list[UserDetail]:
        """Расширенный список (role + is_active) для admin-панели управления
        юзерами. В отличие от list_users_in_org — включает НЕактивных."""
        org = await db.scalar(select(Organization).where(Organization.uid == organization_uid))
        if org is None:
            return []

        users = (
            await db.scalars(
                select(User).where(User.organization_id == org.id).order_by(User.full_name)
            )
        ).all()

        return [
            UserDetail(
                uid=u.uid,
                email=u.email,
                full_name=u.full_name,
                organization_uid=org.uid,
                is_active=u.is_active,
                email_verified_at=u.email_verified_at,
                role=u.role,
            )
            for u in users
        ]


class OrganizationDirectory:
    """Чтение публичных данных организаций."""

    @staticmethod
    async def get(db: AsyncSession, uid: UUID) -> OrganizationPublic | None:
        org = await db.scalar(select(Organization).where(Organization.uid == uid))
        if org is None:
            return None
        return OrganizationPublic.model_validate(org)