"""Strict schema authority for AEAT registry definitions.
Each modelo revision carries an ``output_sensitivity`` field typed as
:class:`SensitivityClass` that governs the encryption tier applied to
generated output envelopes.
"""
from __future__ import annotations
from collections.abc import Callable, Iterable, Mapping, Sequence
from datetime import date
from enum import StrEnum
from typing import Annotated, Final, Literal, TypeGuard
from pydantic import (
BeforeValidator,
Field,
PrivateAttr,
ValidationInfo,
field_validator,
model_validator,
)
from ....core.aggregation import BindingAggregation, BindingSourceKind
from ....core.authority_grade import UNDECLARED_REGISTRY_AUTHORITY_GRADE, RegistryAuthorityGrade
from ....core.casilla_id import CasillaId
from ....core.classification.policies import SensitivityClass
from ....core.errors.hierarchy import pydantic_validation_boundary
from ....core.filing_projection_ref import FilingProjectionRef
from ....core.frozen_mapping import FROZEN_MAPPING
from ....core.modelo import Modelo
from ....core.period import Period, RegistrySelectorPeriodCode
from ....core.revision_review import RevisionReviewStatus
from ....core.tax_domain import TaxDomain
from ....core.type_guards import is_object_mapping
from .binding_provider import BindingProvider
from .binding_temporal import (
AllRevisionContexts,
AuthoredBinding,
BindingApplicability,
BindingAuthorship,
)
from .binding_terminal_origin import TerminalOriginExpectation
from .binding_value_contract import BindingValueChannel, BindingValueContract
from .errors import RegistryValidationError
from .ids import (
ApplicationLinkId,
BindingId,
ConstructId,
CrossReferenceId,
DeadlineWindowId,
DependencyClassificationId,
ExtractionProfileId,
FormulaId,
LegalRefId,
ModeloId,
RevisionId,
SourceRefId,
VerificationExpectationId,
WorkbookParityRefId,
)
from .m303_orden_projection_models import M303AnnualOrdenAuthority
from .period_selector_match import selector_period_matches_request
from .schema_governance import (
validate_attribution_names_somebody,
validate_governance_stamp_coherence,
validate_review_scope,
validate_reviewed_at_within_horizon,
)
from .schema_references import (
RegistryExternalLink,
RegistrySnapshotRef,
TemporalProjectionDirection,
TemporalSupportEnvelope,
)
from .schema_rounding import RegistryRoundingCode as RegistryRoundingCode
from .schema_rounding import RegistryRoundingCodeValue
from .schema_scalars import (
BicString as _BicString,
)
from .schema_scalars import (
CalendarDate as _CalendarDate,
)
from .schema_scalars import (
CCAACode as _CCAACode,
)
from .schema_scalars import (
CountryCode as _CountryCode,
)
from .schema_scalars import DecimalValue as _DecimalValue
from .schema_scalars import (
IbanString as _IbanString,
)
from .schema_scalars import (
ModeloYear as _ModeloYear,
)
from .schema_scalars import (
MunicipalityCode as _MunicipalityCode,
)
from .schema_scalars import (
NifIvaString as _NifIvaString,
)
from .schema_scalars import (
NifString as _NifString,
)
from .schema_scalars import (
PeriodCode as _PeriodCode,
)
from .schema_scalars import (
PersonOrEntityName as _PersonOrEntityName,
)
from .schema_scalars import (
PostalCode as _PostalCode,
)
from .schema_scalars import (
ProvinceCode as _ProvinceCode,
)
from .schema_scalars import (
WorkbookCellRefStr as _WorkbookCellRefStr,
)
from .schema_verification import (
LiveCrossReferenceDecision,
RegistryVerificationPolicy,
VerificationExpectationDefinition,
VerificationPredicateDefinition,
WorkbookParityReference,
fold_reconciliation_total_casilla_ids,
)
__all__ = [
"BindingDefinition",
"DecimalValue",
"FormulaDefinition",
"ModeloDefinition",
"ModeloRevision",
"RegistryCatalogues",
"RegistrySnapshot",
"SnapshotGlobalCatalogues",
"SociedadesAnnualManualCoverageCatalogue",
"SociedadesAnnualManualCoverageDisposition",
"SociedadesAnnualManualCoverageStatus",
"SupportedFilingYearsCatalogue",
]
from ....core.filing_year import FilingYear
from .casilla_structural_succession import CasillaStructuralSuccession, structural_succession_failures
from .cleared_families import ClearedFamilyDeclaration
from .convenio import ConvenioAuthority
from .facts.schema import GovernedFactCatalogue
from .identifier_evolutions import IdentifierEvolution
from .lineage_attestation import LineageAttestation
from .modelo_inception import ModeloInceptionField
from .modelo_localization import require_modelo_localization, resolve_modelo_localization
from .modelo_pending_orden import PendingEjercicioOrdenes
from .restated_families import RestatedFamilyDeclaration
from .revision_contracts import (
DeclaredPredecessor,
RegistryRevisionDeclaration,
validate_revision_predecessors,
)
from .runtime_catalogues import RuntimeRegistryCatalogues
from .schema_base import (
CHAIN_FAMILY,
GOVERNANCE_STAMP,
MANIFEST_ONLY,
SCHEMA_FAMILY,
CalculationClass,
CalculationClassField,
LegalRefs,
ModeloFilingCapabilities,
ModeloFilingCapability,
RegistryAuthorityGradeField,
RegistryModel,
RevisionReviewStatusField,
SensitivityClassField,
SourceCitation,
SourceRefs,
coerce_enum_member,
governance_stamp_fields,
manifest_only_fields,
schema_family_fields,
)
from .schema_deadlines import DeadlineWindowDefinition as _DeadlineWindowDefinition
from .schema_deadlines import ModeloScheduleDefinition as _ModeloScheduleDefinition
from .schema_exports import ExportLayoutDefinition, ProjectionEndpointDeclaration
from .schema_extraction import ExtractionProfileDefinition
from .schema_formula import (
FormulaExpression,
ParameterDefinition,
)
from .schema_references import LegalReference, SourceReference
from .schema_revision_members import (
ApplicabilityRuleDefinition as _ApplicabilityRuleDefinition,
)
from .schema_revision_members import (
ApplicationLinkDefinition as _ApplicationLinkDefinition,
)
from .schema_revision_members import (
ConstructDefinition as _ConstructDefinition,
)
from .schema_revision_members import (
DependencyClassificationDefinition as _DependencyClassificationDefinition,
)
from .schema_surfaces import (
CalculationCompletenessManifest,
CasillaConstraints,
CasillaContinuidadEvolutionDefinition,
CasillaDefinition,
validate_family_identity_uniqueness,
)
# Scalar and annotated value types live in ``_schema_scalars``; retaining these
# assignments keeps the historical ``_schema`` import surface authoritative.
DecimalValue = _DecimalValue
NifString = _NifString
ModeloYear = _ModeloYear
PeriodCode = _PeriodCode
CountryCode = _CountryCode
IbanString = _IbanString
PersonOrEntityName = _PersonOrEntityName
NifIvaString = _NifIvaString
CCAACode = _CCAACode
ProvinceCode = _ProvinceCode
PostalCode = _PostalCode
MunicipalityCode = _MunicipalityCode
BicString = _BicString
CalendarDate = _CalendarDate
WorkbookCellRefStr = _WorkbookCellRefStr
[docs]
class ContinuidadValidationMode(StrEnum):
"""How strictly casilla continuity is validated for a revision."""
ADVISORY = "advisory"
STRICT = "strict"
ContinuidadValidationModeField = Annotated[
ContinuidadValidationMode, BeforeValidator(coerce_enum_member(ContinuidadValidationMode))
]
"""Registry token hydrated into a ContinuidadValidationMode member."""
[docs]
class ModeloCadence(StrEnum):
"""How often a modelo is filed.
``PROFILE_BASED`` is not a period: it says the cadence is decided by the taxpayer's
own circumstances, which is why it cannot be compared against the others as a
frequency.
"""
MONTHLY = "monthly"
QUARTERLY = "quarterly"
ANNUAL = "annual"
AD_HOC = "ad_hoc"
PROFILE_BASED = "profile_based"
ModeloCadenceField = Annotated[ModeloCadence, BeforeValidator(coerce_enum_member(ModeloCadence))]
"""Registry token hydrated into a ModeloCadence member."""
[docs]
class BindingDefinition(RegistryModel):
"""Declare one typed source-to-casilla binding in a registry revision.
The declaration is a closed statement in four parts: :attr:`provider` names
WHERE the value comes from as one member of the discriminated
:data:`~.binding_provider.BindingProvider` union, :attr:`value` states the
typed contract the value honours, :attr:`aggregation` states how several
source facts fold into one, and :attr:`terminal_origins` states the classes
of terminal fact the resolved value is allowed to rest on -- the authored
half of the provenance audit.
The former ``source`` token and untyped ``selector`` mapping are gone: they
were two fields that could disagree, and the union makes the tag part of the
member. :attr:`source` survives as a read-only projection of the member's
discriminator for consumers that still key on the source kind alone.
"""
id: BindingId
provider: BindingProvider
value: BindingValueContract
aggregation: BindingAggregation | None = None
applicability: BindingApplicability = AllRevisionContexts()
terminal_origins: tuple[TerminalOriginExpectation, ...] = ()
authorship: BindingAuthorship = AuthoredBinding()
# AEAT borrador pre-fill tier: the third, AEAT-live prefill tier, distinct
# from the local relation prefill (`_relation_prefill`) and previous-filing
# direct-carry (`_binding_prefill`) tiers. The three share only the word
# "prefill" and must not be merged.
aeat_prefilled: bool = False
legal_refs: LegalRefs
source_refs: SourceRefs
source_citations: tuple[SourceCitation, ...] = Field(default_factory=tuple)
@property
def source(self) -> BindingSourceKind:
"""Return the provider member's discriminator as its source kind.
A projection, never a stored field: the discriminator IS the source
kind, so the two can no longer disagree the way the former sibling
``source`` column could.
"""
return BindingSourceKind(self.provider.kind)
@model_validator(mode="after")
@pydantic_validation_boundary
def _validate_row_set_terminal_cardinality(self) -> BindingDefinition:
"""Refuse a row collection that claims exactly one terminal fact.
A row family rests on however many terminal facts the period produced,
including none: ``exactly_one`` would make an empty row family
indistinguishable from a missing one, which is the collapse
:mod:`binding_terminal_origin`'s cardinality axis exists to prevent.
Whether the rows are grouped or provider-native is the registration's
question; the cardinality is contract-local, so it is checked here.
"""
if self.value.channel is not BindingValueChannel.ROW_SET:
return self
offending = tuple(
expectation.source_class.value
for expectation in self.terminal_origins
if expectation.cardinality == "exactly_one"
)
if offending:
raise RegistryValidationError(
f"binding {self.id!r}: a row_set value cannot rest on an exactly_one terminal origin",
context={"id": self.id, "source_classes": list(offending)},
)
return self
def _frozen_index[K, V](index: Mapping[K, Sequence[V]]) -> dict[K, tuple[V, ...]]:
"""Freeze an accumulating list-valued index into its published immutable form.
Every producer index accumulates into lists rather than overwriting, so a
duplicate declaration stays visible instead of being hidden by a
last-write-wins assignment; freezing is the last step before publication.
"""
return {key: tuple(values) for key, values in index.items()}
[docs]
class SchemaFamilyDispositionDeclaration(RegistryModel):
"""A revision's declared reason that one of its schema families does not apply.
The only way an empty family reads as anything but
:attr:`RegistrySchemaFamilyDisposition.BLOCKED_PENDING_EVIDENCE`, and it is
deliberately expensive to make: a substantive claim about what the law does
not require of this modelo, so it carries a reason somebody wrote and the
references it stands on.
The alternative — an allowlist of families permitted to be empty — was
rejected as the shape of the problem rather than its solution. An allowlist
entry records that somebody wanted the check quiet; this records what they
claim and what backs it, which is the thing a later reviewer can disagree
with.
"""
reason: str = Field(min_length=1, max_length=1024)
legal_refs: LegalRefs
source_refs: SourceRefs
[docs]
class CasillaStorageSelector(RegistryModel):
"""Unambiguous address of one effective casilla in a declared predecessor.
This selector is storage identity only. It neither declares nor implies
legal continuity between the selected member and the successor edition.
"""
revision: RevisionId
id: CasillaId
[docs]
class CasillaFieldOverride(RegistryModel):
"""Typed declaration of fields changed from one predecessor casilla."""
selector: CasillaStorageSelector
fields: Annotated[Mapping[str, object], FROZEN_MAPPING] = Field(default_factory=dict)
removed_fields: tuple[str, ...] = ()
restate_provenance: bool = False
@model_validator(mode="after")
@pydantic_validation_boundary
def _validate_patch(self) -> CasillaFieldOverride:
patchable = set(CasillaDefinition.model_fields) | {"additional_source_refs"}
constraint_patchable = set(CasillaConstraints.model_fields) | {"additional_source_refs"}
malformed = sorted(
path
for path in self.removed_fields
if not path or any(not segment for segment in path.split(".")) or len(path.split(".")) > 2
)
if malformed:
raise RegistryValidationError(f"casilla field override has malformed removed fields {malformed!r}")
unknown_removed = {
path
for path in self.removed_fields
if (
(len(segments := path.split(".")) == 1 and segments[0] not in patchable)
or (len(segments) == 2 and (segments[0] != "constraints" or segments[1] not in constraint_patchable))
)
}
unknown = sorted((set(self.fields) - patchable) | unknown_removed)
if unknown:
raise RegistryValidationError(f"casilla field override names unknown fields {unknown!r}")
overlap = set(self.fields) & set(self.removed_fields)
constraint_fields = self.fields.get("constraints")
for path in self.removed_fields:
segments = path.split(".")
if len(segments) != 2 or segments[0] != "constraints" or "constraints" not in self.fields:
continue
if not isinstance(constraint_fields, Mapping) or segments[1] in constraint_fields:
overlap.add(path)
overlap = sorted(overlap)
if overlap:
raise RegistryValidationError(f"casilla field override both sets and removes fields {overlap!r}")
if not self.fields and not self.removed_fields and not self.restate_provenance:
raise RegistryValidationError("casilla field override must set or remove at least one field")
if "inherited_from" in self.fields or any(
path.split(".")[0] == "inherited_from" for path in self.removed_fields
):
raise RegistryValidationError("inherited_from is loader-owned and cannot be overridden")
return self
[docs]
class CasillaMemberRemoval(RegistryModel):
"""Explicitly remove one member inherited from the declared predecessor."""
selector: CasillaStorageSelector
[docs]
class CasillaMemberPosition(RegistryModel):
"""Place an effective casilla without restating its payload."""
id: CasillaId
position: int = Field(ge=0)
[docs]
class FamilyStorageSelector(RegistryModel):
"""Address one keyed-family member in the immediate predecessor."""
revision: RevisionId
id: str = Field(min_length=1)
[docs]
class FamilyFieldOverride(RegistryModel):
"""Patch only the changed fields of one inherited keyed-family member."""
family: str = Field(min_length=1)
selector: FamilyStorageSelector
fields: Annotated[Mapping[str, object], FROZEN_MAPPING] = Field(default_factory=dict)
removed_fields: tuple[str, ...] = ()
sequence_additions: Annotated[Mapping[str, tuple[object, ...]], FROZEN_MAPPING] = Field(default_factory=dict)
sequence_removals: Annotated[Mapping[str, tuple[int, ...]], FROZEN_MAPPING] = Field(default_factory=dict)
sequence_order: Annotated[Mapping[str, tuple[int, ...]], FROZEN_MAPPING] = Field(default_factory=dict)
restate_identity: bool = False
replacement_id: str | None = Field(default=None, min_length=1)
@model_validator(mode="after")
@pydantic_validation_boundary
def _validate_patch(self) -> FamilyFieldOverride:
from .keyed_families import family_spec
spec = family_spec(self.family)
if spec is None or not spec.keyed:
raise RegistryValidationError(
f"family field override requires an inherited keyed family, got {self.family!r}"
)
if (
not self.fields
and not self.removed_fields
and not self.sequence_additions
and not self.sequence_removals
and not self.sequence_order
and not self.restate_identity
and self.replacement_id is None
):
raise RegistryValidationError("family field override must set or remove at least one field")
overlap = sorted(set(self.fields) & set(self.removed_fields))
if overlap:
raise RegistryValidationError(f"family field override both sets and removes fields {overlap!r}")
if spec.identity in self.fields or spec.identity in self.removed_fields:
raise RegistryValidationError(f"family identity {spec.identity!r} cannot be patched")
sequence_paths = set(self.sequence_additions) | set(self.sequence_removals) | set(self.sequence_order)
if "" in sequence_paths:
raise RegistryValidationError("family sequence paths must not be empty")
for path, indices in (*self.sequence_removals.items(), *self.sequence_order.items()):
if any(index < 0 for index in indices) or len(set(indices)) != len(indices):
raise RegistryValidationError(f"family sequence path {path!r} has invalid indices")
return self
[docs]
class FamilyMemberRemoval(RegistryModel):
"""Explicitly remove one member inherited from a keyed family."""
family: str = Field(min_length=1)
selector: FamilyStorageSelector
[docs]
class FamilyMemberPosition(RegistryModel):
"""Place one effective keyed-family member without restating its payload."""
family: str = Field(min_length=1)
id: str = Field(min_length=1)
position: int = Field(ge=0)
[docs]
class ModeloRevision(RegistryRevisionDeclaration):
"""A single versioned form layout and calculation ruleset for one modelo.
The ``orden_aplicabilidad`` field names the legal-catalogue
:class:`LegalReference` id(s) of the ordenes ministeriales that approve or
amend this revision's form for its declared applicability window
(e.g. ``["orden-hac-277-2026:art-3"]`` for M100 ejercicio 2025).
The field is mandatory at validation time: every revision must cite the
Ordenes that approve or amend the form for its applicability window.
The governance stamp — ``engineered_by``, ``review_status``, ``reviewed_by``,
``reviewed_at``, ``reviewed_against`` — is the revision's *declared*
provenance, optional and fail-closed to
:attr:`RevisionReviewStatus.PENDING_REVIEW` on absence. ``reviewed_against``
is the review's scope on an edition that names a predecessor: the predecessor
the stated rows were reviewed against, required on a reviewed delta edition
and refused everywhere else. Like ``predecessor`` it is excluded from
serialisation when absent, so a revision without it dumps exactly as it did
before the key existed. Its
rules and the reasoning behind them live in :mod:`..schema_governance`,
which the validators below delegate to.
``authority_grade`` is the revision's *declared* authority reach, a separate
subject from the stamp: the stamp says who signed the revision off, the grade
says how far the revision's authority extends. It shares the stamp's
manifest-only placement guarantee — it is a claim about the whole revision,
so it must be readable in ``revision.toml`` rather than merged in from a
fragment thousands deep — and it shares the fail-closed shape, reading as
:data:`~cadrumo.core.authority_grade.UNDECLARED_REGISTRY_AUTHORITY_GRADE` when absent. It is
deliberately optional rather than defaulted on the field, so an ungraded
revision stays distinguishable from one explicitly graded at that same
floor; :attr:`effective_authority_grade` is the reading, and
:attr:`is_graded` the distinction.
``export_layouts`` is the AUTHORED form, not the one that ships. A record
carrying ``binding_record`` is authored thin on purpose -- its envelope
constants only -- and
:func:`~._export.derive_export_layouts_from_bindings` materialises the real
fields at snapshot build. Modelo 369's ``t36904`` is 9 authored fields and
161 derived ones against a design sheet requiring 161; modelo 390's
``page-05`` is 6 and 105. Any consumer comparing a layout against an official
record design MUST resolve through that function, which is the stage the
export-layout record-coverage validation measures when the authority is
built. Reading this attribute for that purpose reports every materialised
field as an unwritten position: it produced 22 confident false
silent-data-loss findings across modelos 369, 390 and 131, twice, in trees
that were already stamped and verified.
``predecessor`` is the revision's explicit declaration of the sibling
edition it is authored relative to. A revision is delta-authored only when
it declares one; nothing infers a predecessor from rows the revision leaves
out, so a revision without the key is a full-copy revision stating every row
itself. Absent reads as ``None`` and is excluded from serialisation, so a
revision that does not declare the key dumps exactly as it did before the
key existed. A :class:`~.revision_contracts.NoPredecessor` is the grounded statement that the
revision chains to no sibling at all, which absence cannot say. Either
declaration is a claim about the whole revision, so it is manifest-only: a
section fragment declaring it is refused. The modelo validates the declared
edges together as a forest through
:func:`~.revision_contracts.validate_predecessor_forest`, and
each edge against the editions' validity dates through
:func:`~.revision_contracts.validate_predecessor_date_agreement`.
``casilla_source_refs`` is the edition's default source grounding for its
casilla rows, declared once rather than restated on every row. It is a
different fact from ``source_refs``, which cites what the edition as a whole
stands on. The loader fills it into every casilla row, and every row's
``constraints`` table, that states no ``source_refs`` of its own. In the same
pass ``orden_aplicabilidad`` fills the ``legal_refs`` of every row and
constraints table that states none, because the edition's approving ordenes
are already declared there once and a second field would duplicate them. A
row or constraints table stating its own value keeps it whole: the default
replaces nothing and is never merged into a stated value, and an explicitly
empty value is refused rather than defaulted. The defaults are applied after
predecessor inheritance, so a row inherited from a predecessor and stating
none takes this edition's defaults, never the predecessor's; source
references are declared per edition. Like ``predecessor`` it is excluded
from serialisation when absent, and manifest-only, since it grounds rows
across every fragment of the edition.
A casilla row or its ``constraints`` table may instead state
``additional_source_refs``: the procedure or form citations that belong to
the box's concept rather than to the edition's design. Such a table's
``source_refs`` is the edition's ``casilla_source_refs`` followed by those
additions, duplicates removed and the default first. The additions travel
with the row, so an inherited row extends the default of the edition it now
sits in. A table stating both keys, additions that are empty, or additions
in an edition declaring no ``casilla_source_refs`` is refused. The key is
consumed by the loader and never reaches this model.
``binding_source_refs`` and ``formula_source_refs`` are the same fact for
the binding and formula families, and carry the same member-side rule: a
row stating no ``source_refs`` takes the edition default, a row stating
``additional_source_refs`` takes the default followed by its additions, and
a row stating ``source_refs`` keeps them whole.
``application_link_source_refs``, ``applicability_source_refs``,
``filing_schedule_source_refs``, ``live_cross_reference_source_refs``,
``extraction_profile_source_refs``,
``dependency_classification_source_refs``, ``construct_source_refs`` and
``parameter_source_refs`` state the same fact, under the same rule, for the
``application_links``, ``applicability``, ``filing_schedules``,
``live_cross_references``, ``extraction_profiles``,
``dependency_classifications``, ``constructs`` and ``parameters``
families.
One field per family rather than one shared field because the families are
grounded in different documents -- a modelo's casillas in its diseno de
registros, its bindings in that design's record layout, its formulas in the
approving orden's instructions, its filing schedules and applicability in
the approving orden's plazos and obligation articles -- and one shared
default would force an edition to restate on every other family whenever
one of them differs. Each is independent: declaring one says nothing about
the others, and an edition declaring none is exactly as it was before these
keys existed. The pairing of family to field is declared once, in
:data:`~.keyed_families.CANONICAL_FAMILY_SPECS`, whose ``source_default_key``
is what the loader fills from, so no family can be defaulted from another's
grounding.
Whether a useful family source default can be derived is an optimization
result calculated from the member declarations. It is not revision data.
"""
localization_key: str = Field(min_length=1, exclude=True, repr=False)
legal_refs: Annotated[LegalRefs, MANIFEST_ONLY]
source_refs: SourceRefs
# Required by validate_orden_aplicabilidad; kept default-empty so the
# validator can report a grounded registry failure instead of a parse error.
orden_aplicabilidad: Annotated[tuple[LegalRefId, ...], MANIFEST_ONLY] = ()
casilla_source_refs: Annotated[SourceRefs | None, MANIFEST_ONLY] = Field(
default=None,
exclude_if=lambda value: value is None,
)
binding_source_refs: Annotated[SourceRefs | None, MANIFEST_ONLY] = Field(
default=None,
exclude_if=lambda value: value is None,
)
formula_source_refs: Annotated[SourceRefs | None, MANIFEST_ONLY] = Field(
default=None,
exclude_if=lambda value: value is None,
)
application_link_source_refs: Annotated[SourceRefs | None, MANIFEST_ONLY] = Field(
default=None,
exclude_if=lambda value: value is None,
)
applicability_source_refs: Annotated[SourceRefs | None, MANIFEST_ONLY] = Field(
default=None,
exclude_if=lambda value: value is None,
)
filing_schedule_source_refs: Annotated[SourceRefs | None, MANIFEST_ONLY] = Field(
default=None,
exclude_if=lambda value: value is None,
)
live_cross_reference_source_refs: Annotated[SourceRefs | None, MANIFEST_ONLY] = Field(
default=None,
exclude_if=lambda value: value is None,
)
extraction_profile_source_refs: Annotated[SourceRefs | None, MANIFEST_ONLY] = Field(
default=None,
exclude_if=lambda value: value is None,
)
dependency_classification_source_refs: Annotated[SourceRefs | None, MANIFEST_ONLY] = Field(
default=None,
exclude_if=lambda value: value is None,
)
construct_source_refs: Annotated[SourceRefs | None, MANIFEST_ONLY] = Field(
default=None,
exclude_if=lambda value: value is None,
)
parameter_source_refs: Annotated[SourceRefs | None, MANIFEST_ONLY] = Field(
default=None,
exclude_if=lambda value: value is None,
)
parameters: Annotated[tuple[ParameterDefinition, ...], SCHEMA_FAMILY] = ()
casillas: Annotated[tuple[CasillaDefinition, ...], SCHEMA_FAMILY] = ()
casilla_storage_baseline: Annotated[RevisionId | None, MANIFEST_ONLY] = Field(
default=None,
exclude_if=lambda value: value is None,
description=(
"Immediate revision whose effective casillas supply storage defaults. "
"This is not a legal-continuity declaration."
),
)
casilla_overrides: Annotated[tuple[CasillaFieldOverride, ...], MANIFEST_ONLY] = Field(
default=(), exclude_if=lambda value: not value
)
casilla_removals: Annotated[tuple[CasillaMemberRemoval, ...], MANIFEST_ONLY] = Field(
default=(), exclude_if=lambda value: not value
)
casilla_positions: Annotated[tuple[CasillaMemberPosition, ...], MANIFEST_ONLY] = Field(
default=(), exclude_if=lambda value: not value
)
family_storage_baseline: Annotated[RevisionId | None, MANIFEST_ONLY] = Field(
default=None,
exclude_if=lambda value: value is None,
description="Immediate revision whose effective keyed families supply storage defaults.",
)
cleared_families: Annotated[tuple[ClearedFamilyDeclaration, ...], MANIFEST_ONLY] = Field(
default=(), exclude_if=lambda value: not value
)
scoped_families: Annotated[tuple[str, ...], MANIFEST_ONLY] = Field(default=(), exclude_if=lambda value: not value)
family_overrides: Annotated[tuple[FamilyFieldOverride, ...], MANIFEST_ONLY] = Field(
default=(), exclude_if=lambda value: not value
)
family_removals: Annotated[tuple[FamilyMemberRemoval, ...], MANIFEST_ONLY] = Field(
default=(), exclude_if=lambda value: not value
)
family_positions: Annotated[tuple[FamilyMemberPosition, ...], MANIFEST_ONLY] = Field(
default=(), exclude_if=lambda value: not value
)
formulas: Annotated[tuple[FormulaDefinition, ...], SCHEMA_FAMILY] = ()
bindings: Annotated[tuple[BindingDefinition, ...], SCHEMA_FAMILY] = ()
projection_endpoints: Annotated[tuple[ProjectionEndpointDeclaration, ...], SCHEMA_FAMILY] = ()
export_layouts: Annotated[tuple[ExportLayoutDefinition, ...], SCHEMA_FAMILY] = ()
extraction_profiles: Annotated[tuple[ExtractionProfileDefinition, ...], SCHEMA_FAMILY] = ()
live_cross_references: Annotated[tuple[LiveCrossReferenceDecision, ...], SCHEMA_FAMILY] = ()
workbook_parity_refs: Annotated[tuple[WorkbookParityReference, ...], SCHEMA_FAMILY] = ()
verification_expectations: Annotated[tuple[VerificationExpectationDefinition, ...], SCHEMA_FAMILY] = ()
application_links: Annotated[tuple[_ApplicationLinkDefinition, ...], SCHEMA_FAMILY] = ()
deadline_windows: Annotated[tuple[_DeadlineWindowDefinition, ...], SCHEMA_FAMILY] = ()
filing_schedules: Annotated[tuple[_ModeloScheduleDefinition, ...], SCHEMA_FAMILY] = ()
constructs: Annotated[tuple[_ConstructDefinition, ...], SCHEMA_FAMILY] = ()
dependency_classifications: Annotated[tuple[_DependencyClassificationDefinition, ...], SCHEMA_FAMILY] = ()
applicability: Annotated[tuple[_ApplicabilityRuleDefinition, ...], SCHEMA_FAMILY] = ()
completeness_manifest: CalculationCompletenessManifest | None = None
verification_predicates: Annotated[tuple[VerificationPredicateDefinition, ...], SCHEMA_FAMILY] = ()
continuidad_validation: ContinuidadValidationModeField = ContinuidadValidationMode.ADVISORY
casilla_continuidad_evolutions: Annotated[tuple[CasillaContinuidadEvolutionDefinition, ...], CHAIN_FAMILY] = ()
casilla_structural_successions: Annotated[tuple[CasillaStructuralSuccession, ...], CHAIN_FAMILY] = ()
identifier_evolutions: Annotated[tuple[IdentifierEvolution, ...], CHAIN_FAMILY] = ()
lineage_attestations: Annotated[tuple[LineageAttestation, ...], MANIFEST_ONLY] = Field(
default=(),
exclude_if=lambda value: not value,
description=(
"Edge-local lineage evidence for members inherited from this revision's declared predecessor. "
"Each attestation names exactly one family identity and is authored in this revision's manifest."
),
)
authority_grade: Annotated[RegistryAuthorityGradeField | None, MANIFEST_ONLY] = None
family_dispositions: Annotated[Mapping[str, SchemaFamilyDispositionDeclaration], MANIFEST_ONLY, FROZEN_MAPPING] = (
Field(default_factory=dict, validate_default=True)
)
restated_families: Annotated[tuple[RestatedFamilyDeclaration, ...], MANIFEST_ONLY] = Field(
default=(),
exclude_if=lambda value: not value,
description=(
"Families this edition states in full, declining inheritance from its predecessor on this edge. "
'Authored in revision.toml as restated_families = [{ family = "formulas", '
'cause = "official_structure_differs", reason = "..." }], one entry per family. The family is '
"closed to the families the merge inherits, the cause to the enumerated restatement causes."
),
)
engineered_by: Annotated[str | None, GOVERNANCE_STAMP] = None
review_status: Annotated[RevisionReviewStatusField, GOVERNANCE_STAMP] = RevisionReviewStatus.PENDING_REVIEW
reviewed_by: Annotated[str | None, GOVERNANCE_STAMP] = None
reviewed_at: Annotated[date | None, GOVERNANCE_STAMP] = None
reviewed_against: Annotated[RevisionId | None, GOVERNANCE_STAMP] = Field(
default=None,
exclude_if=lambda value: value is None,
)
@field_validator("engineered_by", "reviewed_by")
@classmethod
@pydantic_validation_boundary
def _attribution_names_somebody(cls, value: str | None, info: ValidationInfo) -> str | None:
"""Refuse an attribution that is declared but names nobody."""
return validate_attribution_names_somebody(value, field_name=info.field_name)
@field_validator("reviewed_at")
@classmethod
@pydantic_validation_boundary
def _reviewed_at_is_within_the_signoff_horizon(cls, value: date | None) -> date | None:
"""Refuse a signoff date no auditor could ever check."""
return validate_reviewed_at_within_horizon(value)
@property
def is_graded(self) -> bool:
"""Return whether this revision declares an authority grade at all.
The distinction :attr:`effective_authority_grade` deliberately erases: an
ungraded revision and one declared at the floor read as the same scope
but are not the same claim, and only one of them is a backlog entry.
"""
return self.authority_grade is not None
@property
def effective_authority_grade(self) -> RegistryAuthorityGrade:
"""Return the authority reach to act on, reading absence fail-closed.
An undeclared grade reads as
:data:`~cadrumo.core.authority_grade.UNDECLARED_REGISTRY_AUTHORITY_GRADE` — the lowest
rung — so a revision nobody has graded confers scheduling reach and
nothing more. Consumers read the reach here rather than each deciding
for itself what a missing declaration means.
"""
return self.authority_grade if self.authority_grade is not None else UNDECLARED_REGISTRY_AUTHORITY_GRADE
[docs]
def get_label(self, locale: str) -> str | None:
"""Resolve the optional revision label from the shared catalogue."""
return resolve_modelo_localization((self.localization_key,), locale=locale)
@property
def label(self) -> str | None:
"""Return the optional official-Spanish revision label."""
return self.get_label("es")
[docs]
def projection_endpoint_index(self) -> Mapping[FilingProjectionRef, tuple[ProjectionEndpointDeclaration, ...]]:
"""Index declared projection endpoints without concealing duplicates.
Generated layouts deliberately do not participate in this authority:
their fields must later prove an exact bijection with this revision-owned
declaration index.
"""
declarations_by_ref: dict[FilingProjectionRef, list[ProjectionEndpointDeclaration]] = {}
for declaration in self.projection_endpoints:
declarations_by_ref.setdefault(declaration.projection_ref, []).append(declaration)
return _frozen_index(declarations_by_ref)
@model_validator(mode="after")
@pydantic_validation_boundary
def _validate_family_dispositions(self) -> ModeloRevision:
"""Refuse an inapplicability claim that names no family or contradicts one.
Both directions are silent corruption otherwise. A declaration keyed on a
typo names no family, so it resolves nothing while reading as though it
did; and a declaration against a family that HOLDS content asserts the
law does not require what the revision already declares, which is a
contradiction the coverage projection would have to arbitrate.
"""
for family in self.family_dispositions:
if family not in REVISION_SCHEMA_FAMILY_FIELDS:
raise RegistryValidationError(
f"revision {self.id!r} declares a family disposition for {family!r}, which is not a schema "
f"family; enrolled families are {sorted(REVISION_SCHEMA_FAMILY_FIELDS)!r}",
)
if getattr(self, family):
raise RegistryValidationError(
f"revision {self.id!r} declares family {family!r} not applicable but also declares "
f"{len(getattr(self, family))} of them; drop the disposition or drop the content",
)
return self
@model_validator(mode="after")
@pydantic_validation_boundary
def _validate_restated_families(self) -> ModeloRevision:
"""Refuse a restatement claim with nothing behind it or no edge to apply to.
Three directions, each one a declaration that reads as load-bearing while
withdrawing nothing. A family the edition declares EMPTY restates no
members, so the claim would be indistinguishable from a family nobody
has built yet. A root edition inherits nothing on any family, so there is
no edge for the claim to act on. And two entries for one family leave the
merge with two authored reasons for one decision and no rule for picking
between them.
"""
if not self.restated_families:
return self
if not isinstance(self.predecessor, DeclaredPredecessor):
raise RegistryValidationError(
f"revision {self.id!r} declares restated families "
f"{sorted(entry.family for entry in self.restated_families)!r} but has no declared predecessor, "
"so it inherits nothing to decline",
)
seen: set[str] = set()
for entry in self.restated_families:
if entry.family in seen:
raise RegistryValidationError(
f"revision {self.id!r} declares family {entry.family!r} restated twice; one family carries "
"one authored restatement",
)
seen.add(entry.family)
if not getattr(self, entry.family):
raise RegistryValidationError(
f"revision {self.id!r} declares family {entry.family!r} restated in full but declares no "
f"{entry.family!r} at all; drop the restatement or state the family",
)
return self
@model_validator(mode="after")
@pydantic_validation_boundary
def _validate_family_storage_delta(self) -> ModeloRevision:
"""Keep generic storage declarations on one explicit keyed-family vocabulary."""
from .keyed_families import KEYED_FAMILY_SPECS
keyed = {spec.section for spec in KEYED_FAMILY_SPECS}
scoped = {spec.section for spec in KEYED_FAMILY_SPECS if spec.scoped}
cleared = [item.family for item in self.cleared_families]
named = [
*cleared,
*self.scoped_families,
*(item.family for item in self.family_overrides),
*(item.family for item in self.family_removals),
*(item.family for item in self.family_positions),
]
unknown = sorted(set(named) - keyed)
if unknown:
raise RegistryValidationError(f"family storage delta names non-keyed families {unknown!r}")
if named and self.family_storage_baseline is None and self.predecessor is None:
raise RegistryValidationError(
"family storage delta requires family_storage_baseline or a named predecessor"
)
if len(set(cleared)) != len(cleared):
raise RegistryValidationError(
"cleared families must be unique; one family carries one authored clearance, so two entries "
"leave the merge with two reasons for one decision and no rule for picking between them"
)
if len(set(self.scoped_families)) != len(self.scoped_families):
raise RegistryValidationError("scoped families must be unique")
invalid_scopes = sorted(set(self.scoped_families) - scoped)
if invalid_scopes:
raise RegistryValidationError(f"family assertion scope names non-scoped families {invalid_scopes!r}")
operated = {
*(item.family for item in self.family_overrides),
*(item.family for item in self.family_removals),
*(item.family for item in self.family_positions),
}
missing_scopes = sorted((operated & scoped) - set(self.scoped_families))
if missing_scopes:
raise RegistryValidationError(
f"scoped family storage operations require an edition-local assertion scope {missing_scopes!r}"
)
return self
@model_validator(mode="after")
@pydantic_validation_boundary
def _validate_identity_keyed_families(self) -> ModeloRevision:
"""Refuse a revision declaring two members of one keyed family under one id.
The identity is what names the same member across editions, so a
revision holding it twice leaves the inheritance merge with no answer
about which member a successor supersedes or a retirement withdraws.
Refused here, at the revision boundary that owns both families, rather
than resolved downstream by picking one.
"""
validate_family_identity_uniqueness(
"projection_endpoints",
[declaration.id for declaration in self.projection_endpoints],
)
validate_family_identity_uniqueness(
"verification_predicates",
[predicate.id for predicate in self.verification_predicates],
)
return self
@model_validator(mode="after")
@pydantic_validation_boundary
def _validate_governance_stamp(self) -> ModeloRevision:
"""Bind the reviewer identity to the claim that a review happened."""
validate_governance_stamp_coherence(
revision_id=self.id,
review_status=self.review_status,
reviewed_by=self.reviewed_by,
reviewed_at=self.reviewed_at,
)
return self
@model_validator(mode="after")
@pydantic_validation_boundary
def _validate_review_scope(self) -> ModeloRevision:
"""Keep a comparison reference paired with an actual review claim."""
validate_review_scope(
revision_id=self.id,
review_status=self.review_status,
reviewed_against=self.reviewed_against,
)
return self
REVISION_GOVERNANCE_FIELDS: frozenset[str] = governance_stamp_fields(ModeloRevision)
"""The :class:`ModeloRevision` fields that make up the declared governance stamp.
Derived from the :data:`GOVERNANCE_STAMP` marker on the field declarations rather
than hand-listed, and the sole input to the loader's placement refusal. See
:mod:`..schema_governance` for why the stamp must be readable in the manifest
alone and why marking the field is the whole of enrolling it.
This set is the stamp VOCABULARY, narrower than
:data:`REVISION_MANIFEST_ONLY_FIELDS`: it is what the conformance tooling reads
as declared provenance and what the stamp writer emits, so a field pinned to
the manifest for legal-grounding reasons must not appear here.
"""
REVISION_SCHEMA_FAMILY_FIELDS: frozenset[str] = schema_family_fields(ModeloRevision)
"""Every :class:`ModeloRevision` field whose emptiness is a coverage question.
The revision's declared content collections, read back off the
:data:`SCHEMA_FAMILY` markers rather than hand-listed. This is the denominator
of the per-revision coverage manifest: one disposition row per member, always,
so a family nobody has built is a row saying so rather than an absence.
Gated against the collection-shaped fields derived from the model annotations.
"""
REVISION_MANIFEST_ONLY_FIELDS: frozenset[str] = manifest_only_fields(ModeloRevision)
"""Every :class:`ModeloRevision` field that may be declared only in ``revision.toml``.
A superset of :data:`REVISION_GOVERNANCE_FIELDS` by construction, since
:class:`GovernanceStampMarker` is a :class:`ManifestOnlyMarker`. Beyond the
governance stamp it carries the legally load-bearing scalars ``legal_refs``,
``orden_aplicabilidad`` and ``valid_to``, which share the stamp's readability
hazard and raise its stakes, ``casilla_source_refs``, ``binding_source_refs``
and ``formula_source_refs``, which ground rows in
every fragment of the edition, and ``authority_grade``, which is a claim about how
far the whole revision's authority reaches and so belongs in the one file a
reviewer opens; :mod:`..schema_governance` records how a deep
fragment can otherwise supply a revision's legal grounding while
``revision.toml`` reads as though it did not.
"""
MODELO_REVISION_IDS_CONTEXT: Final = "modelo_revision_ids"
"""Validation-context key naming every revision of the modelo a selected view was taken from."""
def _is_object_frozenset(value: object) -> TypeGuard[frozenset[object]]:
"""Narrow an unparameterised runtime frozenset to untrusted entries."""
return isinstance(value, frozenset)
def _context_revision_ids(modelo_id: str, info: ValidationInfo) -> frozenset[str] | None:
"""Return the owning directory's revision identities supplied to a selected-view validation."""
context = info.context
if not is_object_mapping(context) or MODELO_REVISION_IDS_CONTEXT not in context:
return None
declared = context[MODELO_REVISION_IDS_CONTEXT]
if not _is_object_frozenset(declared) or not all(isinstance(item, str) for item in declared):
raise RegistryValidationError(f"modelo {modelo_id!r} revision identity context must be a frozenset of ids")
return frozenset(item for item in declared if isinstance(item, str))
def _validate_selected_view_references(
modelo_id: str,
revisions: Mapping[RevisionId, ModeloRevision],
revision_ids: frozenset[str],
) -> None:
"""Resolve a selected view's cross-revision identity references against its directory.
Invariants that compare payloads of several revisions (predecessor date
agreement and cycles, structural-succession lineage ownership) belong to
complete-modelo validation, which the directory's source modelo passed.
"""
foreign = sorted(set(revisions) - revision_ids)
if foreign:
raise RegistryValidationError(
f"modelo {modelo_id!r} view revisions {foreign!r} are not revisions of its directory; "
f"declared revisions are {sorted(revision_ids)!r}"
)
for revision in revisions.values():
predecessor = revision.predecessor
if isinstance(predecessor, DeclaredPredecessor) and predecessor.revision_id not in revision_ids:
raise RegistryValidationError(
f"modelo {modelo_id!r} revision {revision.id!r} declares predecessor {predecessor.revision_id!r}, "
f"which is not a revision of this modelo; declared revisions are {sorted(revision_ids)!r}"
)
for relation in revision.casilla_structural_successions:
if relation.to_revision != revision.id or relation.from_revision not in revision_ids:
raise RegistryValidationError(
f"structural succession {modelo_id}/{revision.id}/{relation.id}: unknown endpoint revision; "
f"declared revisions are {sorted(revision_ids)!r}"
)
[docs]
class ModeloDefinition(RegistryModel):
"""Declare a modelo and its complete collection of revision authorities."""
id: ModeloId
title_localization_key: str = Field(min_length=1, exclude=True, repr=False)
official_name_localization_key: str = Field(min_length=1, exclude=True, repr=False)
tax_domain: Annotated[TaxDomain, BeforeValidator(lambda v: TaxDomain(v) if isinstance(v, str) else v)]
cadence: ModeloCadenceField
jurisdiction: Literal["ES-AEAT"]
calculation_class: CalculationClassField = CalculationClass.FILING
output_sensitivity: SensitivityClassField = SensitivityClass.FINANCIAL
capabilities: ModeloFilingCapabilities = ()
legal_refs: LegalRefs
source_refs: SourceRefs
inception: Annotated[ModeloInceptionField | None, MANIFEST_ONLY] = Field(
default=None,
exclude_if=lambda value: value is None,
)
pending_ejercicio_ordenes: PendingEjercicioOrdenes = ()
revisions: Annotated[Mapping[RevisionId, ModeloRevision], FROZEN_MAPPING]
# Set only on a directory-selected view: consumer models re-run this
# validator on the same instance without the construction context.
_directory_revision_ids: frozenset[str] | None = PrivateAttr(default=None)
[docs]
def get_title(self, locale: str) -> str:
"""Resolve the Modelo title from the shared catalogue."""
return require_modelo_localization((self.title_localization_key,), locale=locale)
[docs]
def get_official_name(self, locale: str) -> str:
"""Resolve the official Modelo name from the shared catalogue."""
return require_modelo_localization((self.official_name_localization_key,), locale=locale)
@property
def title(self) -> str:
"""Return the strict official-Spanish Modelo title."""
return self.get_title("es")
@property
def official_name(self) -> str:
"""Return the strict official-Spanish Modelo name."""
return self.get_official_name("es")
[docs]
def has_capability(self, name: ModeloFilingCapability) -> bool:
"""Return whether this modelo declares the given capability."""
return name in self.capabilities
@model_validator(mode="after")
@pydantic_validation_boundary
def _validate_revisions(self, info: ValidationInfo) -> ModeloDefinition:
if not self.revisions:
raise RegistryValidationError(f"modelo {self.id!r} must declare at least one revision")
directory_revision_ids = _context_revision_ids(self.id, info)
if directory_revision_ids is None:
directory_revision_ids = self._directory_revision_ids
else:
self._directory_revision_ids = directory_revision_ids
revision_ids = frozenset(self.revisions) if directory_revision_ids is None else directory_revision_ids
for key, revision in self.revisions.items():
if key != revision.id:
raise RegistryValidationError(f"revision key {key!r} does not match revision id {revision.id!r}")
if revision.reviewed_against == revision.id:
raise RegistryValidationError(
f"revision {revision.id!r} has invalid review reference: reviewed_against cannot name itself"
)
if revision.reviewed_against is not None and revision.reviewed_against not in revision_ids:
raise RegistryValidationError(
f"revision {revision.id!r} has dangling review reference "
f"reviewed_against={revision.reviewed_against!r}; declared revisions are "
f"{sorted(revision_ids)!r}"
)
if directory_revision_ids is not None:
_validate_selected_view_references(self.id, self.revisions, directory_revision_ids)
return self
validate_revision_predecessors(self.id, self.revisions)
failures = structural_succession_failures(self)
if failures:
raise RegistryValidationError("; ".join(failures))
return self
def _union_across_expectations[T](
expectations: Sequence[VerificationExpectationDefinition],
select: Callable[[VerificationExpectationDefinition], Iterable[T]],
) -> frozenset[T]:
"""Union one declared set across every expectation folded into a policy.
The fold is a union rather than an intersection on purpose: an id any single
expectation declares is in scope for the snapshot's policy, so a second
expectation cannot narrow the first one's declared coverage away.
"""
return frozenset(value for expectation in expectations for value in select(expectation))
[docs]
class SupportedFilingYearsCatalogue(TemporalSupportEnvelope):
"""The registry's sole declaration of the filing years the product supports.
Authored as bounds rather than an enumeration, because the two ends of the
span do not carry the same force. ``floor`` is a hard gate: nothing resolves
below it. ``horizon`` is the last globally authored coordinate, not a gate;
when ``hard_ceiling`` is absent the supported range remains open above it.
Missing modelo editions inside that range are projected from an eligible
authored anchor.
The span is contiguous by construction. A product that supports 2022 and
2024 but not 2023 is not a state the law produces; the enumerated form could
express it only by accident, and nothing ever did.
:attr:`TemporalSupportEnvelope.years` still enumerates the span, derived
now rather than authored,
so a consumer asking which years are supported keeps asking the same way. It
is a plain property rather than a computed field on purpose: the published
artifact is re-validated against this model on every read, and a serialised
derivation would be refused there as an unexpected member.
"""
[docs]
def admits_filing_year(self, filing_year: int) -> bool:
"""Return whether a filing year is inside the product's hard gates."""
return self.projection_coordinate(filing_year) is not None
[docs]
class SociedadesAnnualManualCoverageStatus(StrEnum):
"""The independently auditable availability state of one annual manual."""
AVAILABLE = "available"
UNACQUIRED = "unacquired"
UNPUBLISHED = "unpublished"
SociedadesAnnualManualCoverageStatusField = Annotated[
SociedadesAnnualManualCoverageStatus,
BeforeValidator(coerce_enum_member(SociedadesAnnualManualCoverageStatus)),
]
"""Registry token hydrated into a Sociedades annual-manual coverage status."""
[docs]
class SociedadesAnnualManualCoverageDisposition(RegistryModel):
"""One exact-year outcome for the annual Sociedades manual corpus.
This is documentary availability only. It is intentionally separate from
Modelo 200 revision selection and filing capability.
"""
year: int = Field(ge=2000, le=2099)
status: SociedadesAnnualManualCoverageStatusField
source_ref: SourceRefId | None = None
official_locator: RegistryExternalLink
observed_at: date
acquisition_condition_key: str | None = Field(default=None, min_length=1)
@model_validator(mode="after")
@pydantic_validation_boundary
def _validate_disposition(self) -> SociedadesAnnualManualCoverageDisposition:
if self.status is SociedadesAnnualManualCoverageStatus.AVAILABLE:
if self.source_ref is None:
raise ValueError("available Sociedades manual coverage requires source_ref")
if self.acquisition_condition_key is not None:
raise ValueError(
"available Sociedades manual coverage must not declare acquisition_condition_key",
)
elif self.status is SociedadesAnnualManualCoverageStatus.UNACQUIRED:
if self.source_ref is not None:
raise ValueError("unacquired Sociedades manual coverage must not declare source_ref")
if self.acquisition_condition_key is None:
raise ValueError(
"unacquired Sociedades manual coverage requires acquisition_condition_key",
)
else:
if self.source_ref is not None:
raise ValueError("unpublished Sociedades manual coverage must not declare source_ref")
if self.acquisition_condition_key is None:
raise ValueError(
"unpublished Sociedades manual coverage requires acquisition_condition_key",
)
return self
[docs]
class SociedadesAnnualManualCoverageCatalogue(RegistryModel):
"""One declarative coverage ledger for annual Sociedades manuals."""
dispositions: tuple[SociedadesAnnualManualCoverageDisposition, ...] = Field(min_length=1)
@field_validator("dispositions")
@classmethod
@pydantic_validation_boundary
def _years_are_unique_and_ordered(
cls,
value: tuple[SociedadesAnnualManualCoverageDisposition, ...],
) -> tuple[SociedadesAnnualManualCoverageDisposition, ...]:
years = tuple(disposition.year for disposition in value)
if tuple(sorted(set(years))) != years:
raise RegistryValidationError(
"Sociedades annual manual coverage years must be unique and in ascending order",
)
return value
[docs]
class RegistryCatalogues(RegistryModel):
"""Collect the registry-wide legal, source, fact, and support catalogues."""
legal: Annotated[Mapping[LegalRefId, LegalReference], FROZEN_MAPPING]
sources: Annotated[Mapping[SourceRefId, SourceReference], FROZEN_MAPPING]
facts: GovernedFactCatalogue = Field(default_factory=GovernedFactCatalogue)
convenio: ConvenioAuthority = Field(default_factory=ConvenioAuthority.empty)
supplementary_ordenes: Annotated[Mapping[Modelo, M303AnnualOrdenAuthority], FROZEN_MAPPING] = Field(
default_factory=dict[Modelo, M303AnnualOrdenAuthority],
validate_default=True,
)
supported_filing_years: SupportedFilingYearsCatalogue | None = None
sociedades_annual_manual_coverage: SociedadesAnnualManualCoverageCatalogue | None = None
runtime: RuntimeRegistryCatalogues = Field(default_factory=RuntimeRegistryCatalogues)
[docs]
def require_supported_filing_years(self) -> SupportedFilingYearsCatalogue:
"""Return the registry's single support envelope, refusing a catalogue set without it."""
if self.supported_filing_years is None:
raise RegistryValidationError("registry catalogues declare no supported filing years")
return self.supported_filing_years
[docs]
class SnapshotGlobalCatalogues(RegistryModel):
"""Small registry-wide values required while assembling a point snapshot."""
convenio: ConvenioAuthority = Field(default_factory=ConvenioAuthority.empty)
supplementary_ordenes: Annotated[Mapping[Modelo, M303AnnualOrdenAuthority], FROZEN_MAPPING] = Field(
default_factory=dict[Modelo, M303AnnualOrdenAuthority],
validate_default=True,
)
supported_filing_years: SupportedFilingYearsCatalogue
[docs]
@classmethod
def from_catalogues(cls, catalogues: RegistryCatalogues) -> SnapshotGlobalCatalogues:
"""Project only globals consumed by snapshot construction and fact resolution."""
return cls(
convenio=catalogues.convenio,
supplementary_ordenes=catalogues.supplementary_ordenes,
supported_filing_years=catalogues.require_supported_filing_years(),
)
[docs]
class RegistrySnapshot(RegistryModel):
"""Represent the resolved immutable authority for one modelo filing coordinate."""
modelo: ModeloDefinition
revision: ModeloRevision
filing_period: Period | None = None
filing_year: FilingYear
authored_filing_year: FilingYear | None = None
supported_filing_years: SupportedFilingYearsCatalogue | None = None
revision_projection_direction: TemporalProjectionDirection = TemporalProjectionDirection.AUTHORED
# Accepts normal period codes and declared event-period names; upstream
# PeriodSelector + ModeloScheduleDefinition constrain the token set.
period: RegistrySelectorPeriodCode
legal: Annotated[Mapping[LegalRefId, LegalReference], FROZEN_MAPPING]
sources: Annotated[Mapping[SourceRefId, SourceReference], FROZEN_MAPPING]
extraction_profiles: Annotated[Mapping[ExtractionProfileId, ExtractionProfileDefinition], FROZEN_MAPPING]
live_cross_references: Annotated[Mapping[CrossReferenceId, LiveCrossReferenceDecision], FROZEN_MAPPING]
workbook_parity_refs: Annotated[Mapping[WorkbookParityRefId, WorkbookParityReference], FROZEN_MAPPING]
verification_expectations: Annotated[
Mapping[VerificationExpectationId, VerificationExpectationDefinition], FROZEN_MAPPING
]
application_links: Annotated[Mapping[ApplicationLinkId, _ApplicationLinkDefinition], FROZEN_MAPPING]
deadline_windows: Annotated[Mapping[DeadlineWindowId, _DeadlineWindowDefinition], FROZEN_MAPPING]
filing_schedules: Annotated[Mapping[str, _ModeloScheduleDefinition], FROZEN_MAPPING]
constructs: Annotated[Mapping[ConstructId, _ConstructDefinition], FROZEN_MAPPING]
dependency_classifications: Annotated[
Mapping[DependencyClassificationId, _DependencyClassificationDefinition], FROZEN_MAPPING
]
convenio: ConvenioAuthority = Field(default_factory=ConvenioAuthority.empty)
supplementary_ordenes: Annotated[Mapping[Modelo, M303AnnualOrdenAuthority], FROZEN_MAPPING] = Field(
default_factory=dict[Modelo, M303AnnualOrdenAuthority],
)
@property
def snapshot_ref(self) -> RegistrySnapshotRef:
"""Return this validated snapshot's canonical persisted coordinate."""
return RegistrySnapshotRef(
modelo=self.modelo.id,
revision_id=self.revision.id,
modelo_year=self.filing_year,
period=self.period,
)
@staticmethod
def _validate_identifier_keyed_map(field_name: str, values: Mapping[str, object]) -> None:
"""Require every snapshot map key to name the payload stored beneath it."""
for key, payload in values.items():
payload_id = getattr(payload, "id", None)
if not isinstance(payload_id, str):
raise RegistryValidationError(
f"snapshot {field_name} payload beneath key {key!r} has no string id",
)
if key != payload_id:
raise RegistryValidationError(
f"snapshot {field_name} key {key!r} does not match payload id {payload_id!r}",
)
@model_validator(mode="after")
@pydantic_validation_boundary
def _validate_identifier_keyed_maps(self) -> RegistrySnapshot:
"""Keep all nested lookup identities aligned with their typed payloads."""
self._validate_identifier_keyed_map("legal", self.legal)
self._validate_identifier_keyed_map("sources", self.sources)
self._validate_identifier_keyed_map("extraction_profiles", self.extraction_profiles)
self._validate_identifier_keyed_map("live_cross_references", self.live_cross_references)
self._validate_identifier_keyed_map("workbook_parity_refs", self.workbook_parity_refs)
self._validate_identifier_keyed_map("verification_expectations", self.verification_expectations)
self._validate_identifier_keyed_map("application_links", self.application_links)
self._validate_identifier_keyed_map("deadline_windows", self.deadline_windows)
self._validate_identifier_keyed_map("filing_schedules", self.filing_schedules)
self._validate_identifier_keyed_map("constructs", self.constructs)
self._validate_identifier_keyed_map("dependency_classifications", self.dependency_classifications)
return self
@model_validator(mode="after")
@pydantic_validation_boundary
def _validate_filing_period_consistency(self) -> RegistrySnapshot:
"""Reconcile :attr:`filing_period` against :attr:`filing_year` and :attr:`period`.
This covers fewer snapshots than its name suggests, and the shortfall is
correct rather than a gap. A snapshot addressed by an administrative censo
coordinate — Modelo 036's ``alta`` / ``modificacion`` / ``baja``, Modelo
145's ``comunicacion`` / ``variacion`` — has no ``filing_period`` to
reconcile, because those coordinates name a registration event rather than
a period a filing occupies and so cannot become a typed ``Period`` at all.
The early return is the only honest answer for them.
Modelo 210's symbolic selector ``EVENT-N`` skips it too, for a different
reason worth separating: it is not an event name but a token standing for
a SET of periods, which the revision matcher expands to the concrete
``EVENT-1`` / ``EVENT-2`` operator scopes. Those concrete scopes DO carry a
filing period and are reconciled normally; only the symbolic form is
skipped, because a set has no single period to check against. Verified
against the registry rather than assumed: the complete skipped set is
M036 ``alta``/``modificacion``/``baja``, M145 ``comunicacion``/``variacion``,
and M210 ``EVENT-N``.
It is stated here because the reduced coverage is invisible at the call
site: nothing about a passing snapshot build reveals that a whole class of
coordinates skipped this check. Do not read a green build as evidence that
every snapshot's filing period was reconciled.
"""
if self.filing_period is None:
return self
if self.filing_period.filing_year != self.filing_year:
raise RegistryValidationError("snapshot filing_period year must match filing_year")
if not selector_period_matches_request(self.period, self.filing_period.registry_token):
raise RegistryValidationError("snapshot filing_period code must match period")
return self
[docs]
def verification_policy(self) -> RegistryVerificationPolicy:
"""Fold this snapshot's verification expectations into one policy.
Returns the registry-grounded :class:`RegistryVerificationPolicy` (union
of computed casilla ids, strictest tolerance, strictest coverage floor).
Raises:
RegistryValidationError: When the snapshot declares no verification
expectations.
"""
expectations = tuple(self.verification_expectations.values())
if not expectations:
raise RegistryValidationError("registry verification requires verification expectations")
return RegistryVerificationPolicy(
expectation_ids=tuple(expectation.id for expectation in expectations),
computed_casilla_ids=_union_across_expectations(
expectations,
lambda expectation: expectation.computed_casilla_ids,
),
reconcile_when_present_casilla_ids=_union_across_expectations(
expectations,
lambda expectation: expectation.reconcile_when_present_casilla_ids,
),
externally_grounded_casilla_ids=_union_across_expectations(
expectations,
lambda expectation: expectation.externally_grounded_casilla_ids,
),
reconciliation_total_casilla_ids=fold_reconciliation_total_casilla_ids(expectations),
tolerance=min(expectation.tolerance for expectation in expectations),
min_coverage=max(expectation.min_coverage for expectation in expectations),
rounding_codes=frozenset(expectation.rounding for expectation in expectations),
discrepancy_causes=_union_across_expectations(
expectations,
lambda expectation: expectation.discrepancy_causes,
),
)