"""Typed read API for the legacy validated-model introspection surfaces.
``RegistryQueryService`` wraps a :class:`ValidatedRegistryAuthority` and exposes
structured report objects for the CLI list, describe, casillas, formulas, and
bindings commands. Queries narrow to a single :class:`ModeloDefinition` and
then to one :class:`ModeloRevision` selected by filing year, period, and
optional revision id.
This service is intentionally not the indexed runtime bridge. Production
operations that hold a :class:`~.authority.PinnedAuthorityOperation` call its
typed ``modelo_directory``/``revision_for_context`` methods directly. The
explicit enumeration methods below remain available only for diagnostic
surfaces whose contract genuinely requires the complete validated graph.
"""
from __future__ import annotations
import re
from collections import Counter, defaultdict
from collections.abc import Mapping
from datetime import date
from decimal import Decimal
from typing import TYPE_CHECKING
from pydantic import BaseModel, ConfigDict, TypeAdapter, ValidationError
from ....core.aggregation import BindingSourceKind
from ....core.authority_grade import RegistryAuthorityGrade
from ....core.i18n.render import output_language
from ....core.modelo import Modelo
from ....core.models import STRICT_FROZEN_CONFIG
from ....core.period import Period, RegistrySelectorPeriodCode
from ....core.tax_domain import TaxDomain
from ....core.type_adapters import OBJECT_TUPLE_ADAPTER
from .authority import ValidatedRegistryAuthority
from .binding_selector_utils import boolean_binding_encoded_values
from .binding_temporal import binding_applies_to_period
from .errors import (
RegistryFailureClassification,
RegistryFailureCondition,
RegistrySnapshotError,
RegistryValidationError,
)
from .ids import BindingId
from .period_selector_match import registry_period_for_request, selector_token_for_request
from .query_reports import (
BindingInputChannel,
ModeloBindingQueryRow,
ModeloBindingsReport,
ModeloCasillaDetailReport,
ModeloCasillaRow,
ModeloCasillasReport,
ModeloDescribeReport,
ModeloFormulaRow,
ModeloFormulasReport,
ModeloListReport,
ModeloListRow,
ModeloSupportMatrixReport,
RegistrySourceInventoryReport,
RegistrySourceInventoryRow,
RegistrySourceSite,
)
from .relation_prefill_bindings import RelationPrefillProvider
from .relations import relation_prefill_bindings_for_period
from .runtime_graph import (
enum_consumed_binding_ids,
expression_binding_refs,
expression_casilla_refs,
expression_parameter_refs,
)
from .schema import BindingDefinition, ModeloDefinition, ModeloRevision
from .schema_base import filing_period_from_scope
from .schema_input_kind import InputKind
from .schema_surfaces import CasillaDefinition
from .support_matrix import build_support_matrix, build_support_matrix_from_directory_views
from .temporal import (
ModeloRevisionDirectory,
select_revision,
select_revision_for_year,
select_revision_metadata_for_year,
)
if TYPE_CHECKING:
from collections.abc import Iterable
from .authority import PinnedAuthorityOperation
from .schema_references import PeriodSelector
_PUBLIC_MAPPING_ADAPTER: TypeAdapter[dict[object, object]] = TypeAdapter(
dict[object, object],
config=ConfigDict(strict=True),
)
#: Bare registry period tokens (``0A``, ``1T``-``4T``, ``01``-``12``,
#: ``1P``-``4P``, ``EXT-1T``-``EXT-4T``, ``AD-HOC``, ``EVENT-N``) carry
#: no filing year. ``describe`` accepts them to narrow a modelo to a
#: revision that declares the token, without forcing the operator to
#: compose an artificial ``YYYY``-prefixed string.
_BARE_PERIOD_RE = re.compile(
r"^(?:0A|[1-4]T|[1-4]P|0[1-9]|1[0-2]|EXT-[1-4]T|AD-HOC|EVENT-\d+)$",
re.I,
)
[docs]
class ResolvedRegistryQueryContext(BaseModel):
"""One resolved registry query context, returned by both resolution forms.
``RegistryQueryService`` reaches a modelo revision by two routes: an
unscoped lookup that narrows by period token alone, and a scoped lookup
that resolves a snapshot for an explicit filing year. Both routes return
this context, so every query method and report builder consumes one typed
shape instead of two positional tuples whose arity and element meaning
differed. The unscoped route leaves :attr:`filing_year` unset, which is
why it stays optional here.
This is deliberately narrower than a
:class:`~domain.calculations.registry.schema.RegistrySnapshot`: a snapshot
requires a filing year and carries the whole legal, source, and
expectation authority, none of which the unscoped period query has or
needs to answer a read-only introspection request.
"""
model_config = STRICT_FROZEN_CONFIG
definition: ModeloDefinition
revision: ModeloRevision
revision_ids: tuple[str, ...] = ()
filing_year: int | None = None
registry_period: RegistrySelectorPeriodCode | None = None
def _latest_revision(definition: ModeloDefinition) -> ModeloRevision:
"""Return the latest revision under the query service's deterministic tie-break."""
return max(definition.revisions.values(), key=lambda item: (item.valid_from, str(item.id)))
def _raise_unscoped_as_of_query(modelo: str) -> None:
"""Refuse a point-in-time argument when no filing year can scope it."""
raise RegistryValidationError(
"as_of point-in-time selection requires a filing-year-scoped query; "
"the unscoped period query resolves the latest revision by period.",
registry_failure=RegistryFailureClassification(
condition=RegistryFailureCondition.QUERY_FILING_YEAR_SCOPED,
facts={
"modelo": modelo.strip(),
"as_of_supplied": True,
"filing_year_supplied": False,
"query_scope": "unscoped_period",
},
),
)
def _period_revision_candidates(
definition: ModeloDefinition,
period: str,
) -> tuple[ModeloRevision, ...]:
"""Return revisions declaring ``period`` in the authority's order."""
return tuple(
revision
for revision in definition.revisions.values()
if selector_token_for_request(revision.period_selector.declared_periods, period) is not None
)
def _resolve_declared_period_revision(
definition: ModeloDefinition,
*,
period: str,
) -> ResolvedRegistryQueryContext:
"""Resolve a bare declared period to its latest declaring revision."""
bare = period.strip()
declared_by_revision = tuple(
token for revision in definition.revisions.values() for token in revision.period_selector.declared_periods
)
token_is_declared = selector_token_for_request(declared_by_revision, bare) is not None
if not (_BARE_PERIOD_RE.fullmatch(bare.upper()) or token_is_declared):
raise RegistryValidationError(
f"period must be a bare registry token; pass the filing year separately; got {period!r}",
)
candidates = _period_revision_candidates(definition, bare)
if not candidates:
declared = sorted(set(declared_by_revision))
raise RegistryValidationError(
f"period {period!r} is not declared by any revision of modelo "
f"{definition.id}; declared periods: {', '.join(declared)}",
)
revision = max(candidates, key=lambda item: (item.valid_from, str(item.id)))
registry_token = selector_token_for_request(revision.period_selector.declared_periods, bare)
if registry_token is None:
raise RegistryValidationError(
f"period {period!r} is not declared by revision {revision.id} of modelo {definition.id}",
)
return ResolvedRegistryQueryContext(
definition=definition,
revision=revision,
registry_period=registry_token,
)
[docs]
class RegistryQueryService:
"""Legacy introspection facade over the eager validated authority graph.
The constructor deliberately accepts only ``ValidatedRegistryAuthority``.
It must not be widened to accept an indexed reader: indexed consumers use
the already-pinned operation directly, keeping eager traversal visible and
preventing a second selection implementation from drifting from the
canonical runtime contract.
"""
def __init__(self, authority: ValidatedRegistryAuthority) -> None:
"""Bind this read-only query service to one validated authority."""
self._authority = authority
[docs]
def modelo_codes(self) -> tuple[str, ...]:
"""Return deterministic modelo identifiers from the authority index.
This is the explicit bulk-enumeration seam for callers that need a
catalogue of model identifiers. It deliberately does not expose the
authority's eager ``modelos`` graph; a subsequent operation must ask
for the revision it actually needs.
"""
return tuple(row.code for row in self.list_modelos().modelos)
[docs]
def revision_for_scope(
self,
modelo: str,
*,
filing_year: int,
period: str,
as_of: date | None = None,
grade: RegistryAuthorityGrade = RegistryAuthorityGrade.APPLICABILITY,
) -> ModeloRevision:
"""Return exactly the revision selected for one explicit operation scope.
The report path owns canonical temporal and grade selection; callers
that need revision payload fields consume this selected result instead
of walking every revision in an authority-wide model graph.
"""
normalized_modelo = modelo.strip()
definition = self._authority.validate_modelo(normalized_modelo)
if grade is not RegistryAuthorityGrade.APPLICABILITY:
return self.resolve_revision_for_scope(
normalized_modelo,
filing_year=filing_year,
period=period,
as_of=as_of,
grade=grade,
).revision
# Metadata and snapshot consumers share the same temporal source.
return select_revision(
definition,
filing_year=filing_year,
period=period,
on=as_of,
support=self._authority.catalogues.supported_filing_years,
)
[docs]
def revision_by_id(self, modelo: str, revision_id: str) -> ModeloRevision:
"""Return one exact revision component by canonical identity."""
normalized = Modelo(modelo).value
revisions = self.iter_modelo_revisions(modelo_codes=(normalized,))
if not revisions:
raise RegistrySnapshotError(
f"modelo {normalized!r} is not present in the calculation registry",
)
for _modelo_id, revision in revisions:
if str(revision.id) == str(revision_id):
return revision
raise RegistryValidationError(
f"modelo {normalized!r} has no revision {revision_id!r}",
)
[docs]
def source_exists(self, source_reference_id: str) -> bool:
"""Return whether one source-reference metadata row is enrolled."""
return source_reference_id in self._authority.catalogues.sources
[docs]
def iter_modelo_revisions(
self,
*,
modelo_codes: tuple[str, ...] | None = None,
) -> tuple[tuple[str, ModeloRevision], ...]:
"""Explicitly enumerate revision components for diagnostics.
Bulk walks are intentionally named and deterministic. They are for
source inventories and other diagnostics that genuinely need every
revision; ordinary runtime consumers should use
:meth:`revision_for_scope`.
"""
selected = self._authority.modelos
if modelo_codes is not None:
allowed = frozenset(Modelo(code).value for code in modelo_codes)
selected = tuple(modelo for modelo in selected if modelo.id in allowed)
return tuple(
(str(modelo.id), revision)
for modelo in sorted(selected, key=lambda item: str(item.id))
for revision in sorted(modelo.revisions.values(), key=lambda item: (item.valid_from, str(item.id)))
)
[docs]
def iter_modelo_definitions(self) -> tuple[ModeloDefinition, ...]:
"""Explicitly enumerate complete model metadata for diagnostics.
This is intentionally separate from point lookup. Support reports and
source inventories are declared bulk operations; ordinary calculation
consumers must use a selected revision method instead.
"""
return tuple(sorted(self._authority.modelos, key=lambda item: str(item.id)))
[docs]
def list_modelos(
self,
*,
year: int | None = None,
domain: TaxDomain | None = None,
) -> ModeloListReport:
"""Return a catalogue listing of all registered modelos.
Each entry is a lightweight ``ModeloListRow`` carrying only summary
fields — no revision details are resolved. The rows are sorted
ascending by modelo code.
Args:
year: When supplied, restricts the listing to modelos that have
at least one revision whose ``period_selector`` covers the
given filing year. ``None`` returns all registered modelos.
domain: When supplied, restricts the listing to modelos whose
registry :class:`~core.tax_domain.TaxDomain` equals the requested
tax family (e.g. ``TaxDomain("iva")``). ``None`` returns every
family. The ``year`` and ``domain`` filters compose: passing
both narrows to modelos that satisfy each.
Returns:
A :class:`~domain.calculations.registry.query_reports.ModeloListReport`
containing the matching rows.
"""
rows = [
ModeloListRow(
code=str(modelo.id),
title=modelo.title,
cadence=modelo.cadence,
tax_domain=modelo.tax_domain,
revision_count=len(modelo.revisions),
)
for modelo in RegistryQueryService(self._authority).iter_modelo_definitions()
if (
year is None
or _selectors_cover_year((revision.period_selector for revision in modelo.revisions.values()), year)
)
and (domain is None or modelo.tax_domain == domain)
]
# Ordered into a pinned local first: pydantic's generated ``__init__``
# accepts a mapping for a nested model, so sorting inline would let that
# permissive parameter type flow back into the lambda and widen ``row``
# to ``ModeloListRow | Mapping[str, Any]``.
ordered: tuple[ModeloListRow, ...] = tuple(sorted(rows, key=lambda row: row.code))
return ModeloListReport(modelos=ordered)
[docs]
def source_inventory(self) -> RegistrySourceInventoryReport:
"""Report every :class:`~core.aggregation.BindingSourceKind` the committed registry declares, and where.
Walks every committed modelo revision and every binding it declares,
grouping by the binding's ``source`` kind. The result records, per
source kind, the committed revisions that declare it and the per-revision
binding count. This is a pure registry introspection surface — it does
not consult the live calculation mesh — so it stays inside the domain
boundary. Application-layer conformance compares this live inventory
directly with executable calculation-route ownership so an unrouted
declaration is refused rather than silently blanked.
Returns:
A :class:`~domain.calculations.registry.query_reports.RegistrySourceInventoryReport`
whose rows are sorted by the source kind's string value; each row's
sites are sorted by ``(modelo, revision_id)``.
"""
sites_by_source: dict[BindingSourceKind, list[RegistrySourceSite]] = defaultdict(list)
for modelo_id, revision in self.iter_modelo_revisions():
counts: Counter[BindingSourceKind] = Counter(binding.source for binding in revision.bindings)
for source, count in counts.items():
sites_by_source[source].append(
RegistrySourceSite(
modelo=modelo_id,
revision_id=str(revision.id),
binding_count=count,
),
)
inventory: list[RegistrySourceInventoryRow] = []
for source, sites in sites_by_source.items():
ordered_sites: tuple[RegistrySourceSite, ...] = tuple(
sorted(sites, key=lambda site: (site.modelo, site.revision_id))
)
inventory.append(
RegistrySourceInventoryRow(
source_kind=source,
sites=ordered_sites,
total_binding_count=sum(site.binding_count for site in ordered_sites),
),
)
ordered_rows: tuple[RegistrySourceInventoryRow, ...] = tuple(
sorted(inventory, key=lambda row: row.source_kind.value)
)
return RegistrySourceInventoryReport(rows=ordered_rows)
[docs]
def support_matrix(self) -> ModeloSupportMatrixReport:
"""Return the registry-wide per-modelo support/capability matrix.
For every modelo the authority can load, builds a :class:`ModeloEntry`
capturing supported revisions, calc/manifest/export/extractor
capability flags, declared per-ejercicio casilla renames, declared
deprecation (support-removal) decisions, and declared AEAT-portal
cross-references — every field read or folded directly from the
loaded registry, never hand-maintained (see
``aeat-calculation-aggregation`` / ``no-silent-under-declaration``).
Returns:
A :class:`~domain.calculations.registry.query_reports.ModeloSupportMatrixReport`
whose entries are sorted by ``modelo_id``.
"""
return ModeloSupportMatrixReport(entries=build_support_matrix(self._authority))
[docs]
def describe_modelo(
self,
modelo: str,
*,
period: str | None = None,
as_of: date | None = None,
) -> ModeloDescribeReport:
"""Return a full describe report for one modelo and its resolved revision.
Resolves the revision using the same precedence logic as the other
query methods: when ``period`` is a bare registry token (e.g. ``"1T"``,
``"0A"``), the revision that declares it is selected; when ``period``
is ``None``, the latest revision by ``valid_from`` is returned. Use
``describe_modelo_for_scope`` when the filing year must participate in
revision selection.
Args:
modelo: Short numeric identifier for the modelo (e.g. ``"303"``).
period: Optional period narrowing. Accepted forms are bare
registry period tokens (``"1T"``, ``"0A"``, ``"01"``-``"12"``)
or declared non-date tokens such as ``"alta"``.
as_of: Optional calendar date for validity gating. Defaults to
today when ``None``.
Returns:
A :class:`~domain.calculations.registry.query_reports.ModeloDescribeReport`
for the resolved revision.
Raises:
``RegistryValidationError``: When ``modelo`` is not registered,
the period is not declared by any revision, or no revision
covers the requested scope.
"""
return _build_modelo_describe_report(self._resolve_revision(modelo, period=period, as_of=as_of))
[docs]
def describe_modelo_for_scope(
self,
modelo: str,
*,
filing_year: int,
period: str,
as_of: date | None = None,
) -> ModeloDescribeReport:
"""Return a :class:`~domain.calculations.registry.query_reports.ModeloDescribeReport` for a scope."""
return _build_modelo_describe_report(
self.resolve_revision_for_scope(modelo, filing_year=filing_year, period=period, as_of=as_of),
)
[docs]
def casillas(
self,
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,
) -> ModeloCasillasReport:
"""Return the casilla (numbered-box) listing for a resolved modelo revision.
A *casilla* is a numbered input box on an AEAT tax form. The listing
includes every casilla in the resolved revision, optionally filtered to
a subset by kind, required flag, or form page number.
Args:
modelo: Short numeric identifier for the modelo (e.g. ``"303"``).
period: Optional period narrowing; see ``describe_modelo`` for
accepted forms.
as_of: Optional calendar date for validity gating.
input_kind: When supplied, restricts rows to casillas of the
given ``InputKind`` (e.g. ``InputKind.MANUAL``,
``InputKind.COMPUTED``).
required: When supplied, restricts rows to casillas whose
``required`` flag matches this value.
form_number: When supplied, restricts rows to casillas on the
given physical form page or sub-form.
Returns:
A :class:`~domain.calculations.registry.query_reports.ModeloCasillasReport`
for the resolved revision, containing the filtered casilla rows.
Raises:
``RegistryValidationError``: When the modelo or period is not
registered, or no revision covers the requested scope.
"""
return _build_modelo_casillas_report(
self._resolve_revision(modelo, period=period, as_of=as_of),
input_kind=input_kind,
required=required,
form_number=form_number,
)
[docs]
def casillas_for_scope(
self,
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,
) -> ModeloCasillasReport:
"""Return a :class:`~domain.calculations.registry.query_reports.ModeloCasillasReport` for a scope."""
return _build_modelo_casillas_report(
self.resolve_revision_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 casilla(
self,
modelo: str,
casilla: str,
*,
period: str | None = None,
as_of: date | None = None,
) -> ModeloCasillaDetailReport:
"""Return the full semantic detail for one casilla on a resolved revision.
Addresses a single casilla by its canonical id or its printed
``number`` and surfaces the authoritative label, legal/source
grounding, input kind, and — when the casilla is computed — the
resolved formula expression. Revision selection follows the same
precedence as :meth:`describe_modelo`.
Args:
modelo: Short numeric identifier for the modelo (e.g. ``"303"``).
casilla: Casilla id or printed number to look up.
period: Optional period narrowing; see :meth:`describe_modelo`.
as_of: Optional calendar date for validity gating.
Returns:
A :class:`~domain.calculations.registry.query_reports.ModeloCasillaDetailReport`
for the addressed casilla.
Raises:
``RegistryValidationError``: When the modelo or period is not
registered, no revision covers the requested scope, or the
casilla id/number is not defined by the resolved revision.
"""
return _casilla_detail_report(self._resolve_revision(modelo, period=period, as_of=as_of), casilla)
[docs]
def casilla_for_scope(
self,
modelo: str,
casilla: str,
*,
filing_year: int,
period: str,
as_of: date | None = None,
) -> ModeloCasillaDetailReport:
"""Return a :class:`~domain.calculations.registry.query_reports.ModeloCasillaDetailReport` for a scope."""
return _casilla_detail_report(
self.resolve_revision_for_scope(modelo, filing_year=filing_year, period=period, as_of=as_of),
casilla,
)
[docs]
def bindings_for_scope(
self,
modelo: str,
*,
filing_year: int,
period: str,
as_of: date | None = None,
) -> ModeloBindingsReport:
"""Return bindings for a specific filing scope (already-parsed year + period).
Unlike `bindings`, this method accepts the already-parsed ``filing_year``
integer and registry ``period`` string (e.g. ``"1T"``, ``"01"``) produced
by the CLI's period-parsing step. This avoids re-parsing a user-facing
period string when the caller already holds the decomposed values.
Returns:
A :class:`~domain.calculations.registry.query_reports.ModeloBindingsReport`
for the requested filing scope.
"""
return _build_modelo_bindings_report(
self.resolve_revision_for_scope(modelo, filing_year=filing_year, period=period, as_of=as_of),
)
[docs]
def bindings_for_year(
self,
modelo: str,
*,
filing_year: int,
as_of: date | None = None,
) -> ModeloBindingsReport:
"""Return revision-wide bindings for one filing year and effective date."""
return _build_modelo_bindings_report(
self._resolve_revision_for_year(modelo, filing_year=filing_year, as_of=as_of),
)
[docs]
def bindings(
self,
modelo: str,
*,
period: str | None = None,
as_of: date | None = None,
) -> ModeloBindingsReport:
"""Return the full binding listing for a resolved modelo revision.
A *binding* maps a financial-data source to a casilla or formula
input. For year-specific binding ids (e.g. when a multi-revision
modelo publishes different binding names per renta year) prefer
``bindings_for_scope`` so the resolved revision matches the one
the calculation engine will use.
Args:
modelo: Short numeric identifier for the modelo (e.g. ``"130"``).
period: Optional period narrowing; see ``describe_modelo`` for
accepted forms. When ``None``, the latest revision is used.
as_of: Optional calendar date for validity gating.
Returns:
A :class:`~domain.calculations.registry.query_reports.ModeloBindingsReport`
for the resolved revision.
Raises:
``RegistryValidationError``: When the modelo or period is not
registered, or no revision covers the requested scope.
"""
return _build_modelo_bindings_report(self._resolve_revision(modelo, period=period, as_of=as_of))
def _resolve_revision(
self,
modelo: str,
*,
period: str | None,
as_of: date | None,
) -> ResolvedRegistryQueryContext:
if as_of is not None:
_raise_unscoped_as_of_query(modelo)
normalized_modelo = modelo.strip()
definition = self._authority.validate_modelo(normalized_modelo)
revision_ids = self._revision_ids(normalized_modelo)
if period is None:
return ResolvedRegistryQueryContext(
definition=definition,
revision=_latest_revision(definition),
revision_ids=revision_ids,
)
context = _resolve_declared_period_revision(definition, period=period)
return context.model_copy(update={"revision_ids": revision_ids})
[docs]
def resolve_revision_for_scope(
self,
modelo: str,
*,
filing_year: int,
period: str,
as_of: date | None,
grade: RegistryAuthorityGrade = RegistryAuthorityGrade.APPLICABILITY,
) -> ResolvedRegistryQueryContext:
"""Resolve the unique revision governing one explicit filing scope."""
normalized_modelo = modelo.strip()
definition = self._authority.validate_modelo(normalized_modelo)
revision_ids = self._revision_ids(normalized_modelo)
requested_period = period.strip()
declared_by_revision = tuple(
token for revision in definition.revisions.values() for token in revision.period_selector.declared_periods
)
registry_period = (
registry_period_for_request(declared_by_revision, requested_period) or requested_period.upper()
)
# Introspection, not a filing assertion. The FILING rung additionally
# demands a reviewed revision, a reviewed legal set and an export
# layout, and this resolver feeds only read reports -- describe,
# casillas, formulas, bindings. Building at the default rung made every
# one of them refuse an applicability-grade modelo: `describe 036
# --period ALTA` raised "declares no export layout" while answering a
# question about which revision governs an event kind. Every structural
# check -- legal applicability, reference identity, the revision-scoped
# legal and source windows -- runs at this rung too, so the reports lose
# nothing they read.
snapshot = self._authority.snapshot(
str(definition.id),
filing_year=filing_year,
period=registry_period,
on=as_of,
grade=grade,
)
return ResolvedRegistryQueryContext(
definition=definition,
revision=snapshot.revision,
revision_ids=revision_ids,
filing_year=filing_year,
registry_period=registry_period,
)
def _resolve_revision_for_year(
self,
modelo: str,
*,
filing_year: int,
as_of: date | None,
) -> ResolvedRegistryQueryContext:
normalized_modelo = modelo.strip()
definition = self._authority.validate_modelo(normalized_modelo)
revision_ids = self._revision_ids(normalized_modelo)
revision = select_revision_for_year(
definition,
filing_year=filing_year,
on=as_of,
support=self._authority.catalogues.supported_filing_years,
)
year_periods = revision.period_selector.periods_for_year(filing_year)
if not year_periods:
raise RegistryValidationError(
f"modelo {definition.id} revision {revision.id!r} has no period token for filing year {filing_year}",
)
return ResolvedRegistryQueryContext(
definition=definition,
revision=revision,
revision_ids=revision_ids,
filing_year=filing_year,
registry_period=year_periods[0],
)
def _revision_ids(self, modelo: str) -> tuple[str, ...]:
"""Return explicit revision metadata for a model-wide report."""
return tuple(str(revision.id) for _modelo_id, revision in self.iter_modelo_revisions(modelo_codes=(modelo,)))
[docs]
class PinnedRegistryQueryService:
"""Report facade over one generation-pinned component operation.
The legacy :class:`RegistryQueryService` remains the explicit eager
diagnostic facade. Application discovery uses this point-loaded variant:
selector-complete directories provide metadata and each report hydrates
only the revision selected for that query.
"""
def __init__(self, operation: PinnedAuthorityOperation) -> None:
"""Bind the facade to one already-leased authority operation."""
self._operation = operation
[docs]
def modelo_codes(self) -> tuple[str, ...]:
"""Return deterministic model identities from the pinned index."""
return self._operation.modelo_ids()
[docs]
def iter_modelo_revisions(
self,
*,
modelo_codes: tuple[str, ...] | None = None,
) -> tuple[tuple[str, ModeloRevision], ...]:
"""Explicitly enumerate point-loaded revisions for diagnostics."""
allowed = None if modelo_codes is None else frozenset(Modelo(code).value for code in modelo_codes)
rows: list[tuple[str, ModeloRevision]] = []
for modelo_id in sorted(self._operation.modelo_ids()):
if allowed is not None and modelo_id not in allowed:
continue
directory = self._operation.modelo_directory(modelo_id)
for metadata in directory.revisions:
rows.append((modelo_id, self._operation.revision(modelo_id, str(metadata.id))))
return tuple(rows)
def _latest_modelo_views(self) -> tuple[tuple[ModeloRevisionDirectory, ModeloDefinition], ...]:
"""Pair each modelo directory with the view materialized around its latest revision."""
views: list[tuple[ModeloRevisionDirectory, ModeloDefinition]] = []
for modelo_id in sorted(self._operation.modelo_ids()):
directory = self._operation.modelo_directory(modelo_id)
metadata = max(directory.revisions, key=lambda item: (item.valid_from, str(item.id)))
views.append(
(directory, directory.materialize(self._operation.revision(modelo_id, str(metadata.id)))),
)
return tuple(views)
[docs]
def iter_modelo_definitions(self) -> tuple[ModeloDefinition, ...]:
"""Materialize one latest revision view per modelo for diagnostics."""
return tuple(definition for _directory, definition in self._latest_modelo_views())
[docs]
def list_modelos(
self,
*,
year: int | None = None,
domain: TaxDomain | None = None,
) -> ModeloListReport:
"""Return a metadata catalogue read from the modelo directories alone.
Every listed field is modelo-level metadata the directory carries, and
revision count and year coverage come from its complete revision
metadata, so no revision payload is decoded.
"""
rows: list[ModeloListRow] = []
for modelo_id in self._operation.modelo_ids():
directory = self._operation.modelo_directory(modelo_id)
modelo = directory.modelo
selectors = (metadata.period_selector for metadata in directory.revisions)
if year is not None and not _selectors_cover_year(selectors, year):
continue
if domain is not None and TaxDomain(modelo.tax_domain) != domain:
continue
rows.append(
ModeloListRow(
code=modelo.id,
title=modelo.title,
cadence=modelo.cadence,
tax_domain=TaxDomain(modelo.tax_domain),
revision_count=len(directory.revisions),
),
)
return ModeloListReport(modelos=tuple(sorted(rows, key=lambda row: row.code)))
[docs]
def support_matrix(self) -> ModeloSupportMatrixReport:
"""Build the explicit bulk support inventory from point-loaded views."""
return ModeloSupportMatrixReport(entries=build_support_matrix_from_directory_views(self._latest_modelo_views()))
[docs]
def describe_modelo(
self,
modelo: str,
*,
period: str | None = None,
as_of: date | None = None,
) -> ModeloDescribeReport:
"""Return one unscoped modelo description report."""
return _build_modelo_describe_report(self._resolve_revision(modelo, period=period, as_of=as_of))
[docs]
def describe_modelo_for_scope(
self,
modelo: str,
*,
filing_year: int,
period: str,
as_of: date | None = None,
) -> ModeloDescribeReport:
"""Return one filing-year-scoped modelo description report."""
return _build_modelo_describe_report(
self.resolve_revision_for_scope(modelo, filing_year=filing_year, period=period, as_of=as_of),
)
[docs]
def casillas(
self,
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,
) -> ModeloCasillasReport:
"""Return the filtered casilla report for one unscoped modelo query."""
return _build_modelo_casillas_report(
self._resolve_revision(modelo, period=period, as_of=as_of),
input_kind=input_kind,
required=required,
form_number=form_number,
)
[docs]
def casillas_for_scope(
self,
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,
) -> ModeloCasillasReport:
"""Return the filtered casilla report for one filing-year scope."""
return _build_modelo_casillas_report(
self.resolve_revision_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 casilla(
self,
modelo: str,
casilla: str,
*,
period: str | None = None,
as_of: date | None = None,
) -> ModeloCasillaDetailReport:
"""Return one casilla detail from an unscoped modelo query."""
return _casilla_detail_report(self._resolve_revision(modelo, period=period, as_of=as_of), casilla)
[docs]
def casilla_for_scope(
self,
modelo: str,
casilla: str,
*,
filing_year: int,
period: str,
as_of: date | None = None,
) -> ModeloCasillaDetailReport:
"""Return one casilla detail from a filing-year scope."""
return _casilla_detail_report(
self.resolve_revision_for_scope(modelo, filing_year=filing_year, period=period, as_of=as_of),
casilla,
)
[docs]
def bindings(
self,
modelo: str,
*,
period: str | None = None,
as_of: date | None = None,
) -> ModeloBindingsReport:
"""Return bindings for one unscoped modelo query."""
return _build_modelo_bindings_report(self._resolve_revision(modelo, period=period, as_of=as_of))
[docs]
def bindings_for_year(
self,
modelo: str,
*,
filing_year: int,
as_of: date | None = None,
) -> ModeloBindingsReport:
"""Return bindings for one filing year."""
return _build_modelo_bindings_report(
self._resolve_revision_for_year(modelo, filing_year=filing_year, as_of=as_of),
)
[docs]
def bindings_for_scope(
self,
modelo: str,
*,
filing_year: int,
period: str,
as_of: date | None = None,
) -> ModeloBindingsReport:
"""Return bindings for one filing-year scope."""
return _build_modelo_bindings_report(
self.resolve_revision_for_scope(modelo, filing_year=filing_year, period=period, as_of=as_of),
)
def _revision_ids(self, directory: ModeloRevisionDirectory) -> tuple[str, ...]:
return tuple(str(metadata.id) for metadata in directory.revisions)
def _context(
self,
directory: ModeloRevisionDirectory,
revision: ModeloRevision,
*,
filing_year: int | None = None,
registry_period: RegistrySelectorPeriodCode | None = None,
) -> ResolvedRegistryQueryContext:
return ResolvedRegistryQueryContext(
definition=directory.materialize(revision),
revision=revision,
revision_ids=self._revision_ids(directory),
filing_year=filing_year,
registry_period=registry_period,
)
def _resolve_revision(
self,
modelo: str,
*,
period: str | None,
as_of: date | None,
) -> ResolvedRegistryQueryContext:
if as_of is not None:
_raise_unscoped_as_of_query(modelo)
directory = self._operation.modelo_directory(modelo.strip())
if period is None:
metadata = max(directory.revisions, key=lambda item: (item.valid_from, str(item.id)))
return self._context(directory, self._operation.revision(directory.modelo_id, str(metadata.id)))
requested = period.strip()
declared = tuple(token for item in directory.revisions for token in item.period_selector.declared_periods)
if not (_BARE_PERIOD_RE.fullmatch(requested.upper()) or selector_token_for_request(declared, requested)):
raise RegistryValidationError(
f"period must be a bare registry token; pass the filing year separately; got {period!r}",
)
candidates = tuple(
item
for item in directory.revisions
if selector_token_for_request(item.period_selector.declared_periods, requested) is not None
)
if not candidates:
raise RegistryValidationError(
f"period {period!r} is not declared by any revision of modelo {directory.modelo_id}; "
f"declared periods: {', '.join(sorted(set(declared)))}",
)
metadata = max(candidates, key=lambda item: (item.valid_from, str(item.id)))
registry_period = selector_token_for_request(metadata.period_selector.declared_periods, requested)
if registry_period is None:
raise RegistryValidationError(
f"period {period!r} is not declared by revision {metadata.id} of modelo {directory.modelo_id}",
)
return self._context(
directory,
self._operation.revision(directory.modelo_id, str(metadata.id)),
registry_period=registry_period,
)
[docs]
def resolve_revision_for_scope(
self,
modelo: str,
*,
filing_year: int,
period: str,
as_of: date | None,
) -> ResolvedRegistryQueryContext:
"""Resolve the unique revision governing one explicit filing scope."""
directory = self._operation.modelo_directory(modelo.strip())
requested = period.strip()
declared = tuple(token for item in directory.revisions for token in item.period_selector.declared_periods)
registry_period = registry_period_for_request(declared, requested) or requested.upper()
revision = self._operation.revision_for_context(
directory.modelo_id,
filing_year=filing_year,
period=registry_period,
on=as_of,
)
selected_token = selector_token_for_request(revision.period_selector.declared_periods, requested)
return self._context(
directory,
revision,
filing_year=filing_year,
registry_period=selected_token or registry_period,
)
def _resolve_revision_for_year(
self,
modelo: str,
*,
filing_year: int,
as_of: date | None,
) -> ResolvedRegistryQueryContext:
directory = self._operation.modelo_directory(modelo.strip())
# The same canonical year selection every other consumer uses, so a
# year-scoped report names the revision a snapshot of that year selects.
metadata = select_revision_metadata_for_year(directory, filing_year=filing_year, on=as_of)
revision = self._operation.revision(directory.modelo_id, str(metadata.id))
return self._context(
directory,
revision,
filing_year=filing_year,
registry_period=metadata.period_selector.periods_for_year(filing_year)[0],
)
def _build_modelo_describe_report(context: ResolvedRegistryQueryContext) -> ModeloDescribeReport:
"""Assemble a :class:`ModeloDescribeReport` from a resolved query context."""
definition = context.definition
revision = context.revision
filing_year = context.filing_year
registry_period = context.registry_period
return ModeloDescribeReport(
code=str(definition.id),
title=definition.get_title(output_language()),
official_name=definition.get_official_name(output_language()),
tax_domain=definition.tax_domain,
cadence=definition.cadence,
jurisdiction=definition.jurisdiction,
revision=str(revision.id),
revision_ids=context.revision_ids,
filing_year=filing_year,
filing_period=_query_filing_period(filing_year, registry_period),
period=registry_period,
valid_from=revision.valid_from,
valid_to=revision.valid_to,
periods=(
revision.period_selector.declared_periods
if filing_year is None
else revision.period_selector.periods_for_year(filing_year)
),
casilla_count=len(revision.casillas),
manual_casilla_count=sum(1 for casilla in revision.casillas if casilla.input_kind == InputKind.MANUAL),
bound_casilla_count=sum(1 for casilla in revision.casillas if casilla.input_kind == InputKind.BOUND),
computed_casilla_count=sum(1 for casilla in revision.casillas if casilla.input_kind == InputKind.COMPUTED),
binding_count=len(revision.bindings),
formula_count=len(revision.formulas),
legal_refs=tuple(str(ref) for ref in revision.legal_refs),
source_refs=tuple(str(ref) for ref in revision.source_refs),
)
def _casilla_row_included(
casilla: CasillaDefinition,
*,
input_kind: InputKind | None,
required: bool | None,
form_number: str | None,
) -> bool:
"""Return whether a casilla passes the report's optional kind/required/form filters."""
return (
(input_kind is None or casilla.input_kind == input_kind)
and (required is None or casilla.required is required)
and (form_number is None or casilla.form_number == form_number)
)
def _build_modelo_casillas_report(
context: ResolvedRegistryQueryContext,
*,
input_kind: InputKind | None,
required: bool | None,
form_number: str | None,
) -> ModeloCasillasReport:
"""Assemble a filtered :class:`ModeloCasillasReport` from a resolved query context."""
definition = context.definition
revision = context.revision
rows = [
ModeloCasillaRow(
casilla_id=casilla.id,
number=casilla.number,
label=casilla.get_label(output_language()),
help_text=casilla.get_help(output_language()),
section=tuple(casilla.section),
data_type=casilla.data_type,
input_kind=casilla.input_kind,
required=casilla.required,
formula=str(casilla.formula) if casilla.formula is not None else None,
binding=str(casilla.binding) if casilla.binding is not None else None,
form_number=casilla.form_number,
legal_refs=tuple(str(ref) for ref in casilla.legal_refs),
source_refs=tuple(str(ref) for ref in casilla.source_refs),
)
for casilla in revision.casillas
if _casilla_row_included(casilla, input_kind=input_kind, required=required, form_number=form_number)
]
return ModeloCasillasReport(
code=str(definition.id),
revision=str(revision.id),
filing_year=context.filing_year,
filing_period=_query_filing_period(context.filing_year, context.registry_period),
period=context.registry_period,
rows=tuple(rows),
)
def _build_modelo_formulas_report(context: ResolvedRegistryQueryContext) -> ModeloFormulasReport:
"""Assemble a :class:`ModeloFormulasReport` from a resolved query context."""
definition = context.definition
revision = context.revision
fold_binding_ids = frozenset(binding.id for binding, _ in relation_prefill_bindings_for_period(revision))
rows = tuple(
ModeloFormulaRow(
formula_id=str(formula.id),
target_casilla_id=formula.target_casilla_id,
input_casilla_ids=tuple(dict.fromkeys(expression_casilla_refs(formula.expression))),
input_bindings=tuple(dict.fromkeys(expression_binding_refs(formula.expression))),
input_parameters=tuple(dict.fromkeys(expression_parameter_refs(formula.expression))),
input_relations=tuple(
binding_id
for binding_id in dict.fromkeys(expression_binding_refs(formula.expression))
if binding_id in fold_binding_ids
),
expression=_public_mapping(formula.expression.model_dump(mode="json")),
legal_refs=tuple(str(ref) for ref in formula.legal_refs),
source_refs=tuple(str(ref) for ref in formula.source_refs),
)
for formula in revision.formulas
)
return ModeloFormulasReport(
code=str(definition.id),
revision=str(revision.id),
filing_year=context.filing_year,
filing_period=_query_filing_period(context.filing_year, context.registry_period),
period=context.registry_period,
rows=rows,
)
def _build_modelo_bindings_report(context: ResolvedRegistryQueryContext) -> ModeloBindingsReport:
"""Assemble a :class:`ModeloBindingsReport` from a resolved query context."""
modelo = str(context.definition.id)
return ModeloBindingsReport(
code=modelo,
revision=str(context.revision.id),
filing_year=context.filing_year,
filing_period=_query_filing_period(context.filing_year, context.registry_period),
period=context.registry_period,
rows=_binding_rows(context.revision, modelo=modelo, period=context.registry_period),
)
def _binding_rows(
revision: ModeloRevision,
*,
modelo: str | None = None,
period: str | None = None,
) -> tuple[ModeloBindingQueryRow, ...]:
"""Build the typed binding rows for one revision.
Shared by every ``bindings*`` query so the operator-facing
``input_channel`` discriminator is computed once, consistently:
the channel is ``enum`` only for bindings a dispatch op consumes
as a string enum key, ``decimal`` for every other binding.
"""
enum_consumed = enum_consumed_binding_ids(revision)
fold_binding_ids = frozenset(
binding.id for binding, _ in relation_prefill_bindings_for_period(revision, period=period)
)
operator_required = _operator_input_required_by_binding(revision, modelo=modelo, period=period)
return tuple(
ModeloBindingQueryRow(
binding_id=binding.id,
typed_enum=binding.value.typed_enum,
input_channel=(BindingInputChannel.ENUM if binding.id in enum_consumed else BindingInputChannel.DECIMAL),
provider=binding.provider,
aggregation={"op": binding.aggregation.op.value} if binding.aggregation is not None else None,
legal_refs=tuple(binding.legal_refs),
source_refs=tuple(binding.source_refs),
borrador_capable=binding.aeat_prefilled is True,
relation_inputs=(binding.id,) if binding.id in fold_binding_ids else (),
encoded_options=boolean_binding_encoded_values(binding),
operator_input_required=operator_required.get(binding.id, True),
)
for binding in revision.bindings
)
def _find_casilla_for_detail(
context: ResolvedRegistryQueryContext,
casilla: str,
) -> CasillaDefinition:
"""Resolve a detail query by canonical id first, then by printed number."""
definition = context.definition
revision = context.revision
needle = casilla.strip()
matched = next(
(item for item in revision.casillas if str(item.id) == needle or item.number == needle),
None,
)
if matched is not None:
return matched
valid_ids = [str(item.id) for item in revision.casillas]
sample = ", ".join(valid_ids[:20])
overflow = "" if len(valid_ids) <= 20 else f" (+{len(valid_ids) - 20} more)"
raise RegistryValidationError(
f"casilla {casilla!r} is not defined by revision {revision.id} of modelo {definition.id}; "
f"valid casilla ids include: {sample}{overflow}.",
registry_failure=RegistryFailureClassification(
condition=RegistryFailureCondition.QUERY_CASILLA_DECLARED,
facts={
"modelo": str(definition.id),
"revision": str(revision.id),
"casilla": needle,
"casilla_declared": False,
},
),
)
def _casilla_formula_expression(
revision: ModeloRevision,
casilla: CasillaDefinition,
) -> Mapping[str, object] | None:
"""Return a computed casilla's public formula expression, when present."""
if casilla.formula is None:
return None
formula = next((item for item in revision.formulas if item.id == casilla.formula), None)
if formula is None:
return None
return _public_mapping(formula.expression.model_dump(mode="json"))
def _casilla_detail_report(context: ResolvedRegistryQueryContext, casilla: str) -> ModeloCasillaDetailReport:
"""Build the single-casilla detail report, resolving the formula expression.
The casilla is matched by canonical id first, then by printed ``number``
(the same dual key the ``casillas --number`` filter accepts). An unknown
casilla raises a :class:`RegistryValidationError` with its observed
condition and a bounded sample of valid ids.
A computed casilla's ``formula`` id is resolved against the revision's
formulas so the structured expression rides the report.
"""
definition = context.definition
revision = context.revision
matched = _find_casilla_for_detail(context, casilla)
formula_expression = _casilla_formula_expression(revision, matched)
return ModeloCasillaDetailReport(
code=str(definition.id),
revision=str(revision.id),
filing_year=context.filing_year,
filing_period=_query_filing_period(context.filing_year, context.registry_period),
period=context.registry_period,
casilla_id=matched.id,
number=matched.number,
label=matched.get_label(output_language()),
help_text=matched.get_help(output_language()),
section=tuple(matched.section),
data_type=matched.data_type,
input_kind=matched.input_kind,
required=matched.required,
legal_refs=tuple(str(ref) for ref in matched.legal_refs),
source_refs=tuple(str(ref) for ref in matched.source_refs),
binding=matched.binding,
formula_id=matched.formula,
formula_expression=formula_expression,
)
def _relation_prefill_is_period_default(
binding: BindingDefinition,
*,
modelo: str,
period: str,
) -> bool:
"""Return whether a relation-prefill slot is a same-model prior default here.
The slot is fed by this modelo's own earlier periods and does not apply to
the target period, so the operator is not expected to supply it: the value
either arrives from a prior filing or the slot is legitimately out of scope.
"""
provider = binding.provider
if not isinstance(provider, RelationPrefillProvider):
return False
if binding_applies_to_period(binding.applicability, period):
return False
return provider.relation_kind == "previous_period" and str(provider.source_modelo) == modelo
def _operator_input_required_by_binding(
revision: ModeloRevision,
*,
modelo: str | None,
period: str | None,
) -> dict[BindingId, bool]:
"""Return missing-input visibility for relation slots with period-scoped defaults."""
required = {binding.id: True for binding in revision.bindings}
if modelo is None or modelo != Modelo("202").value or period is None:
return required
for binding in revision.bindings:
if _relation_prefill_is_period_default(binding, modelo=modelo, period=period):
required[binding.id] = False
return required
def _selectors_cover_year(selectors: Iterable[PeriodSelector], year: int) -> bool:
return any(selector.includes_year(year) for selector in selectors)
def _query_filing_period(filing_year: int | None, period: str | None) -> Period | None:
if filing_year is None or period is None:
return None
return filing_period_from_scope(filing_year, period)
def _public_mapping(value: object) -> dict[str, object]:
return {str(key): _public_value(item) for key, item in _public_mapping_items(value).items()}
def _public_mapping_items(value: object) -> dict[object, object]:
"""Validate an untyped mapping before projecting it onto the public query API."""
mapping = _try_public_mapping(value)
if mapping is not None:
return mapping
raise RegistryValidationError(f"unsupported public mapping value {value!r}")
def _try_public_mapping(value: object) -> dict[object, object] | None:
"""Return a checked public mapping, or ``None`` when the value is not one."""
if not isinstance(value, Mapping):
return None
try:
return _PUBLIC_MAPPING_ADAPTER.validate_python(value)
except ValidationError:
return None
def _public_tuple(value: object) -> tuple[object, ...] | None:
"""Narrow a runtime tuple to object entries for recursive public projection."""
if not isinstance(value, tuple):
return None
return OBJECT_TUPLE_ADAPTER.validate_python(value)
def _public_value(value: object) -> object:
if isinstance(value, Decimal):
return format(value, "f")
tuple_value = _public_tuple(value)
if tuple_value is not None:
return tuple(_public_value(item) for item in tuple_value)
mapping = _try_public_mapping(value)
if mapping is not None:
return {str(key): _public_value(item) for key, item in mapping.items()}
return value
__all__ = [
"PinnedRegistryQueryService",
"RegistryQueryService",
"ResolvedRegistryQueryContext",
]