DataOps-platform/app/core/knowledge/access.py

from __future__ import annotations

from collections.abc import Iterable
from dataclasses import dataclass

from sqlalchemy import text

from app.core.system.permissions import permissions_for_roles


@dataclass(frozen=True)
class KnowledgeAccessContext:
    subject_id: str
    roles: frozenset[str]
    permissions: frozenset[str]
    business_domain_uids: frozenset[str]
    correlation_id: str
    global_access: bool = False

    def permits_domain(self, business_domain_uid: str | None) -> bool:
        return self.global_access or (
            business_domain_uid is not None
            and business_domain_uid in self.business_domain_uids
        )


def build_access_context(
    session,
    *,
    identity: dict,
    requested_business_domains: Iterable[str] | None,
    correlation_id: str,
) -> KnowledgeAccessContext:
    roles = frozenset(str(role) for role in identity.get("roles", ()))
    global_access = "admin" in roles
    grants: frozenset[str]
    if global_access:
        grants = frozenset()
    else:
        rows = (
            session.execute(
                text(
                    """
                SELECT CAST(business_domain_uid AS text)
                FROM public.user_business_domain_grants
                WHERE user_id = CAST(:user_id AS uuid)
                  AND grant_type IN ('read', 'manage')
                """
                ),
                {"user_id": identity["id"]},
            )
            .scalars()
            .all()
        )
        grants = frozenset(str(value) for value in rows)
    if requested_business_domains is not None:
        requested = frozenset(str(value) for value in requested_business_domains)
        grants = requested if global_access else grants & requested
        global_access = False
    return KnowledgeAccessContext(
        subject_id=str(identity["id"]),
        roles=roles,
        permissions=permissions_for_roles(roles),
        business_domain_uids=grants,
        correlation_id=correlation_id,
        global_access=global_access,
    )