Source code for cadrumo.application.aggregation.source_profile

"""Source-mesh resolver for user-profile backed registry bindings.

Owns :attr:`~core.aggregation.BindingSourceKind.PROFILE` and returns the
profile-sourced values as a :class:`~.source_mesh.CalculationSourceResolution`.
Accepts an optional
:class:`~domain.calculations.registry.schema.RegistrySnapshot` at construction;
when none is supplied the resolver fetches the matching snapshot lazily from the
resident registry authority at resolution time.
"""

from __future__ import annotations

from collections.abc import Iterable
from typing import ClassVar

from ...core.aggregation import BindingSourceKind
from ...domain.calculations.registry.authority import PinnedAuthorityOperation
from ...domain.calculations.registry.ids import BindingId
from ...domain.calculations.registry.schema import RegistrySnapshot
from .source_mesh import (
    CalculationSourceContext,
    CalculationSourceResolution,
)




class ProfileSourceResolver:
    """Resolve calculation-relevant ``source = "profile"`` bindings."""

    resolver_id: ClassVar[str] = "profile"
    owned_sources: ClassVar[tuple[BindingSourceKind, ...]] = (BindingSourceKind.PROFILE,)

    def __init__(
        self,
        *,
        caller_binding_ids: Iterable[BindingId] = (),
        registry_snapshot: RegistrySnapshot | None = None,
        profile_record: object | None = None,
        operation: PinnedAuthorityOperation,
    ) -> None:
        """Capture the caller-owned operation and source-precedence inputs."""
        self._caller_binding_ids = frozenset(caller_binding_ids)
        self._registry_snapshot = registry_snapshot
        self._profile_record = profile_record
        self._operation = operation



    def resolve(self, context: CalculationSourceContext) -> CalculationSourceResolution:
        """Resolve all ``source="profile"`` bindings for the given calculation context.

        Fetches the :class:`RegistrySnapshot` lazily from the resident
        registry authority when none was supplied at construction, then
        delegates to ``resolve_profile_sourced_bindings``, which returns the
        complete :class:`CalculationSourceResolution` (typed binding channels
        plus a :class:`CalculationSourceProvenance` row per profile-sourced
        binding).

        Args:
            context: A :class:`CalculationSourceContext` carrying the modelo,
                filing year, period, and bucket identifier needed to select
                the correct registry snapshot.

        Returns:
            A :class:`CalculationSourceResolution` with typed binding maps
            and a provenance tuple covering every profile-sourced binding.
        """
        snapshot = self._registry_snapshot
        if snapshot is None:
            snapshot = self._operation.snapshot(
                context.modelo,
                filing_year=context.filing_year,
                period=context.period.registry_token,
            )

        from ..modelo.profile_binding import resolve_profile_sourced_bindings

        profile_record = self._profile_record
        if profile_record is None and context.profile is not None:
            profile_record = context.profile.record
        return resolve_profile_sourced_bindings(
            snapshot,
            bucket_id=context.bucket_id,
            profile_record=profile_record,
            caller_binding_ids=self._caller_binding_ids,
            operation=self._operation,
        )




__all__ = ["ProfileSourceResolver"]