Source code for cadrumo.application.modelo.registry_discovery

"""Application Modelo registry discovery queries.

CLI discovery commands call this module instead of constructing the eager
legacy query service or reading the registry authority directly. Each query
leases one indexed operation and returns the domain query report unchanged.

The ``*_for_scope`` helpers accept a concrete :class:`~core.period.Period` and
pass its filing year plus bare registry token into the query service, so
revision selection stays inside the authority-backed registry layer.
"""

from __future__ import annotations

from collections.abc import Callable
from datetime import date

from ...core.period import Period
from ...core.tax_domain import TaxDomain
from ...domain.calculations.registry.authority import PinnedAuthorityOperation
from ...domain.calculations.registry.errors import RegistryValidationError
from ...domain.calculations.registry.queries import PinnedRegistryQueryService
from ...domain.calculations.registry.query_reports import (
    ModeloBindingsReport,
    ModeloCasillaDetailReport,
    ModeloCasillasReport,
    ModeloDescribeReport,
    ModeloFormulasReport,
    ModeloListReport,
    ModeloSupportMatrixReport,
)
from ...domain.calculations.registry.schema_input_kind import InputKind


def _with_service[DiscoveryResult](
    operation: PinnedAuthorityOperation,
    callback: Callable[[PinnedRegistryQueryService], DiscoveryResult],
) -> DiscoveryResult:
    """Run one discovery query inside the caller-owned pinned operation."""
    return callback(PinnedRegistryQueryService(operation))


def _refuse_unscoped_as_of(*, as_of: date | None, scoped_form: str) -> None:
    """Refuse an as_of argument on an unscoped discovery query, naming the scoped form.

    The unscoped path resolves the latest revision by period and cannot gate a
    point-in-time ``as_of`` against a revision's validity window; honouring it
    needs a filing year. Rather than accept-and-silently-ignore the argument
    (the accepted-parameter lie this contract closes), refuse it here at the
    operator-facing boundary and name the scoped query that does honour it.
    """
    if as_of is not None:
        raise RegistryValidationError(
            f"as_of is only honoured by the filing-year-scoped registry query; the unscoped "
            f"query resolves the latest revision by period and cannot apply a historical "
            f"boundary. Re-run {scoped_form} with an explicit filing year to honour as_of.",
        )


[docs] def declared_modelo_period_tokens( modelo: str | None, *, operation: PinnedAuthorityOperation, ) -> tuple[str, ...]: """Return every period token declared by any revision of one modelo.""" if not modelo or not modelo.strip(): return () return _with_service( operation, lambda service: tuple( sorted( { token for _modelo_id, revision in service.iter_modelo_revisions(modelo_codes=(modelo.strip(),)) for token in revision.period_selector.declared_periods }, ), ), )
[docs] def registry_modelo_codes(*, operation: PinnedAuthorityOperation) -> tuple[str, ...]: """Return registry-backed modelo codes in authority order.""" return _with_service(operation, lambda service: tuple(row.code for row in service.list_modelos().modelos))
[docs] def registry_support_matrix(*, operation: PinnedAuthorityOperation) -> ModeloSupportMatrixReport: """Return the registry-wide per-modelo support/capability matrix. Every :class:`~domain.calculations.registry.support_matrix.ModeloEntry` is derived directly from the loaded registry authority (calc-grade, manifest, export formats, extractor, declared casilla renames, declared deprecation decisions, and declared AEAT-portal cross-references) — never hand-maintained. """ return _with_service(operation, lambda service: service.support_matrix())
[docs] def registry_list_modelos( *, year: int | None = None, domain: TaxDomain | None = None, operation: PinnedAuthorityOperation, ) -> ModeloListReport: """Return the registry modelo list report, optionally filtered by tax family.""" return _with_service(operation, lambda service: service.list_modelos(year=year, domain=domain))
[docs] def registry_describe_modelo( modelo: str, *, period: str | None = None, as_of: date | None = None, operation: PinnedAuthorityOperation, ) -> ModeloDescribeReport: """Return the registry modelo description report.""" _refuse_unscoped_as_of(as_of=as_of, scoped_form="registry_describe_modelo_for_scope") return _with_service(operation, lambda service: service.describe_modelo(modelo, period=period, as_of=as_of))
[docs] def registry_describe_modelo_for_scope( modelo: str, *, period: Period, as_of: date | None = None, operation: PinnedAuthorityOperation, ) -> ModeloDescribeReport: """Return the registry modelo description report for an exact :class:`Period`.""" return _with_service( operation, lambda service: service.describe_modelo_for_scope( modelo, filing_year=period.filing_year, period=period.registry_token, as_of=as_of, ), )
[docs] def registry_describe_modelo_for_registry_scope( modelo: str, *, filing_year: int, period: str, as_of: date | None = None, operation: PinnedAuthorityOperation, ) -> ModeloDescribeReport: """Return a modelo description for an explicit filing-year registry scope.""" return _with_service( operation, lambda service: service.describe_modelo_for_scope( modelo, filing_year=filing_year, period=period, as_of=as_of, ), )
[docs] def registry_casillas( modelo: str, *, period: str | None = None, as_of: date | None = None, input_kind: InputKind | None = None, required: bool | None = None, form_number: str | None = None, operation: PinnedAuthorityOperation, ) -> ModeloCasillasReport: """Return the registry casilla report.""" _refuse_unscoped_as_of(as_of=as_of, scoped_form="registry_casillas_for_scope") return _with_service( operation, lambda service: service.casillas( modelo, period=period, as_of=as_of, input_kind=input_kind, required=required, form_number=form_number, ), )
[docs] def registry_casillas_for_scope( modelo: str, *, period: Period, as_of: date | None = None, input_kind: InputKind | None = None, required: bool | None = None, form_number: str | None = None, operation: PinnedAuthorityOperation, ) -> ModeloCasillasReport: """Return the registry casilla report for an exact :class:`Period`.""" return _with_service( operation, lambda service: service.casillas_for_scope( modelo, filing_year=period.filing_year, period=period.registry_token, as_of=as_of, input_kind=input_kind, required=required, form_number=form_number, ), )
[docs] def registry_casillas_for_registry_scope( modelo: str, *, filing_year: int, period: str, as_of: date | None = None, input_kind: InputKind | None = None, required: bool | None = None, form_number: str | None = None, operation: PinnedAuthorityOperation, ) -> ModeloCasillasReport: """Return casillas for an explicit filing-year registry scope.""" return _with_service( operation, lambda service: service.casillas_for_scope( modelo, filing_year=filing_year, period=period, as_of=as_of, input_kind=input_kind, required=required, form_number=form_number, ), )
[docs] def registry_casilla( modelo: str, casilla: str, *, period: str | None = None, as_of: date | None = None, operation: PinnedAuthorityOperation, ) -> ModeloCasillaDetailReport: """Return the single-casilla semantic detail report.""" _refuse_unscoped_as_of(as_of=as_of, scoped_form="registry_casilla_for_registry_scope") return _with_service( operation, lambda service: service.casilla(modelo, casilla, period=period, as_of=as_of), )
[docs] def registry_casilla_for_registry_scope( modelo: str, casilla: str, *, filing_year: int, period: str, as_of: date | None = None, operation: PinnedAuthorityOperation, ) -> ModeloCasillaDetailReport: """Return one casilla detail for an explicit filing-year registry scope.""" return _with_service( operation, lambda service: service.casilla_for_scope( modelo, casilla, filing_year=filing_year, period=period, as_of=as_of, ), )
[docs] def registry_bindings( modelo: str, *, period: str | None = None, as_of: date | None = None, operation: PinnedAuthorityOperation, ) -> ModeloBindingsReport: """Return the registry bindings report for a modelo and optional period.""" _refuse_unscoped_as_of(as_of=as_of, scoped_form="registry_bindings_for_year") return _with_service(operation, lambda service: service.bindings(modelo, period=period, as_of=as_of))
[docs] def registry_bindings_for_year( modelo: str, *, filing_year: int, as_of: date | None = None, operation: PinnedAuthorityOperation, ) -> ModeloBindingsReport: """Return the registry bindings report for a filing year.""" return _with_service( operation, lambda service: service.bindings_for_year(modelo, filing_year=filing_year, as_of=as_of), )
[docs] def registry_bindings_for_scope( modelo: str, *, period: Period, as_of: date | None = None, operation: PinnedAuthorityOperation, ) -> ModeloBindingsReport: """Return the registry bindings report for an exact :class:`Period`.""" return _with_service( operation, lambda service: service.bindings_for_scope( modelo, filing_year=period.filing_year, period=period.registry_token, as_of=as_of, ), )
[docs] def registry_formulas( modelo: str, *, period: str | None = None, as_of: date | None = None, operation: PinnedAuthorityOperation, ) -> ModeloFormulasReport: """Return the registry formulas report.""" _refuse_unscoped_as_of(as_of=as_of, scoped_form="registry_formulas_for_scope") return _with_service(operation, lambda service: service.formulas(modelo, period=period, as_of=as_of))
[docs] def registry_formulas_for_scope( modelo: str, *, period: Period, as_of: date | None = None, operation: PinnedAuthorityOperation, ) -> ModeloFormulasReport: """Return the registry formulas report for an exact :class:`Period`.""" return _with_service( operation, lambda service: service.formulas_for_scope( modelo, filing_year=period.filing_year, period=period.registry_token, as_of=as_of, ), )
[docs] def registry_formulas_for_registry_scope( modelo: str, *, filing_year: int, period: str, as_of: date | None = None, operation: PinnedAuthorityOperation, ) -> ModeloFormulasReport: """Return formulas for an explicit filing-year registry scope.""" return _with_service( operation, lambda service: service.formulas_for_scope( modelo, filing_year=filing_year, period=period, as_of=as_of, ), )
__all__ = [ "declared_modelo_period_tokens", "registry_bindings", "registry_bindings_for_scope", "registry_bindings_for_year", "registry_casilla", "registry_casilla_for_registry_scope", "registry_casillas", "registry_casillas_for_registry_scope", "registry_casillas_for_scope", "registry_describe_modelo", "registry_describe_modelo_for_registry_scope", "registry_describe_modelo_for_scope", "registry_formulas", "registry_formulas_for_registry_scope", "registry_formulas_for_scope", "registry_list_modelos", "registry_modelo_codes", "registry_support_matrix", ]