cadrumo.domain.calculations.registry.authority module

Generation-pinned access to compiled registry authority components.

IndexedRegistryAuthority is the production boundary. It admits the descriptor-selected SQLite generation and loads typed components through one leased PinnedAuthorityOperation. The eager ValidatedRegistryAuthority remains only for development validation and the paired pre-cutover JSON benchmark.

Core types: ModeloRevision.

class RegistryAuthorityCapture(projection, comparison_domain, generation)[source]

Bases: object

One isolated law-selected registry projection and its currentness coordinate.

Parameters:
  • projection (RegistryAuthorityProjection)

  • comparison_domain (ContentDigest)

  • generation (int)

projection: TypeAliasType
comparison_domain: TypeAliasType
generation: int
require_current(current)[source]

Require this immutable artifact capture to match its current coordinate.

Return type:

RegistryAuthorityCapture

Parameters:

current (RegistryAuthorityCurrentCoordinate)

class RegistryAuthorityCurrentCoordinate(comparison_domain, generation)[source]

Bases: object

Opaque same-process coordinate for one registry authority owner scope.

Parameters:
comparison_domain: TypeAliasType
generation: int
require_current(captured)[source]

Require a capture from this exact artifact/process coordinate.

Return type:

RegistryAuthorityCurrentCoordinate

Parameters:

captured (RegistryAuthorityCapture)

class ValidatedRegistryAuthority(modelos, catalogues, _modelos_by_id, _snapshots, _identity_digest='', evidence=<factory>, _profile_schema=None)[source]

Bases: object

Load, validate, and cache registry material behind one access point.

Parameters:
modelos: tuple[ModeloDefinition, ...]
catalogues: RegistryCatalogues
evidence: AuthorityEvidenceProjection
classmethod from_validated_components(*, modelos, catalogues, identity_digest, evidence=None, profile_schema=None)[source]

Construct an immutable runtime projection after development validation.

This boundary accepts models, catalogues, identity, and optional published evidence only. It has no authoring-root or validator input.

Core types: ModeloDefinition.

Return type:

ValidatedRegistryAuthority

Parameters:
profile_schema()[source]

Return the profile declaration captured in this authority generation.

Return type:

ProfileSchemaDefinition

profile_create_context()[source]

Bind development profile creation to this immutable compiled generation.

Return type:

ProfileCreateContext

profile_decode_context()[source]

Bind development secure decode to this immutable compiled generation.

Return type:

ProfileDecodeContext

modelo(modelo_id)[source]

Return a modelo definition by id.

Return type:

ModeloDefinition

Returns:

The ModeloDefinition for modelo_id.

Parameters:

modelo_id (str | Modelo)

supported_filing_years()[source]

Return the registry’s single filing-year support envelope.

Return type:

SupportedFilingYearsCatalogue

project_filing_year(filing_year)[source]

Project an admitted filing year onto the authority’s authored horizon.

Return type:

int

Parameters:

filing_year (int)

tax_domain(tax_domain, *, effective_date=None)[source]

Return a syntax-valid tax domain only when this authority declares it.

Return type:

TaxDomain

Parameters:
resolve_governed_fact(query)[source]

Resolve one typed governed-fact query through this validated authority.

Return type:

ResolvedScalarFact | ResolvedBracketFact | ResolvedMappingFact | ResolvedEntitySetFact | ResolvedOverrideFact | ResolvedEventFact | ResolvedMultiOutputFact

Parameters:

query (Annotated[ScalarFactQuery | BracketFactQuery | MappingFactQuery | EntitySetFactQuery | OverrideFactQuery | EventFactQuery | MultiOutputFactQuery, FieldInfo(annotation=NoneType, required=True, discriminator='family')])

validate_modelo(modelo_id)[source]

Validate one modelo once and return its definition.

Return type:

ModeloDefinition

Returns:

The validated ModeloDefinition for modelo_id.

Parameters:

modelo_id (str)

inspect_revision(modelo_id, *, filing_year, period, on=None)[source]

Project static admission facts for one canonically selected revision.

The request coordinate selects one revision through the canonical temporal resolver, but the returned projection retains neither filing year nor period. It validates the full registry and never constructs a RegistrySnapshot; callers that need filing eligibility must use snapshot() instead.

Return type:

RegistryRevisionInspection

Parameters:
  • modelo_id (str)

  • filing_year (int)

  • period (str)

  • on (date | None)

validate_registry()[source]

Assert that this already-published authority is available.

Return type:

None

snapshot(modelo_id, *, filing_year, period, on=None, revision_id=None, grade=RegistryAuthorityGrade.FILING)[source]

Return the shared immutable snapshot for one admitted filing context.

grade names the rung of authority the CALLER needs and defaults to the strictest one, so a caller that says nothing is unchanged. It exists because this accessor had no way to ask for a lower rung: it always built at FILING, so a modelo whose registry declares authority_grade = applicability – modelo 036, whose censal alta/modificacion/baja is filed on AEAT’s sede and produces no fichero here – refused every caller that only wanted to know which revision governs an event kind.

The rung is part of the cache key. Without it a snapshot built for one rung would be served to a caller asking for another, which is precisely the silent capability claim the grade exists to prevent.

Return type:

RegistrySnapshot

Parameters:
capture_law_selected_projection(modelo_id, *, filing_year, period, on=None, grade=None)[source]

Atomically capture a law-selected inspection or grade-admitted snapshot.

grade=None deliberately selects the static-inspection authority; supplying a grade selects the existing snapshot admission path. The returned projection is deeply immutable and belongs to this generation.

Return type:

RegistryAuthorityCapture

Parameters:
read_current_coordinate()[source]

Return the typed current coordinate for same-domain capture validation.

Return type:

RegistryAuthorityCurrentCoordinate

deadline_windows(year, *, modelos=None)[source]

Return canonically owned, validated deadline windows for year.

Window rows retain their containing revision for provenance, but that containment does not choose the governing revision. The filing coordinate does, through the same select_revision() authority used by snapshots. Non-owning historical copies are therefore never projected, including when a fingerprint-certified warm load predates a stricter corpus validation verdict. This is selection, not deduplication: every row in the selected revision remains observable.

Return type:

tuple[tuple[str, ModeloRevision, DeadlineWindowDefinition], ...]

Parameters:
class PinnedAuthorityOperation(_reader, generation)[source]

Bases: object

Typed component access confined to one leased authority generation.

Parameters:
generation: AuthorityGenerationPin
pin()[source]

Return this operation’s already-leased generation pin.

Return type:

AuthorityGenerationPin

load(query, *, pin)[source]

Load an addressed component only through this operation’s generation.

Return type:

object

Parameters:
governed_fact(fact_id)[source]

Load one raw governed fact by canonical identity.

Return type:

GovernedFact

Parameters:

fact_id (str)

snapshot_globals()[source]

Load the registry-wide globals published with this generation.

Return type:

SnapshotGlobalCatalogues

supported_filing_years()[source]

Return the generation’s single filing-year support envelope.

Return type:

SupportedFilingYearsCatalogue

profile_schema(schema_id='cadrumo.user_profile')[source]

Load the profile declaration used by this exact operation generation.

Return type:

ProfileSchemaDefinition

Parameters:

schema_id (str)

profile_create_context()[source]

Return the required context for constructing new taxpayer profile values.

Return type:

ProfileCreateContext

profile_decode_context()[source]

Return the required context for decoding encrypted taxpayer profile values.

Return type:

ProfileDecodeContext

revision(modelo_id, revision_id)[source]

Load one typed base revision without separately addressed export layouts.

Return type:

ModeloRevision

Parameters:
revision_with_export_layouts(modelo_id, revision_id)[source]

Compose one revision with only its separately addressed export layouts.

Return type:

ModeloRevision

Parameters:
modelo_directory(modelo_id)[source]

Load the small selector-complete directory for one modelo.

Return type:

ModeloRevisionDirectory

Parameters:

modelo_id (str | Modelo)

modelo_ids()[source]

Return deterministic modelo identities without hydrating directories.

Return type:

tuple[str, ...]

revision_ids()[source]

Return deterministic modelo/revision identities without hydrating payloads.

Return type:

tuple[tuple[str, str], ...]

revision_for_context(modelo_id, *, filing_year, period, on=None, revision_id=None)[source]

Select by complete metadata, then load exactly one complete revision.

Return type:

ModeloRevision

Parameters:
  • modelo_id (str | Modelo)

  • filing_year (int)

  • period (str)

  • on (date | None)

  • revision_id (RevisionId | None)

snapshot(modelo_id, *, filing_year, period, on=None, revision_id=None, grade=RegistryAuthorityGrade.FILING)[source]

Build one validated snapshot from same-generation point components.

Return type:

RegistrySnapshot

Parameters:
resolve_governed_fact(query)[source]

Resolve one fact once per query without revalidating its static component.

Return type:

ResolvedScalarFact | ResolvedBracketFact | ResolvedMappingFact | ResolvedEntitySetFact | ResolvedOverrideFact | ResolvedEventFact | ResolvedMultiOutputFact

Parameters:

query (Annotated[ScalarFactQuery | BracketFactQuery | MappingFactQuery | EntitySetFactQuery | OverrideFactQuery | EventFactQuery | MultiOutputFactQuery, FieldInfo(annotation=NoneType, required=True, discriminator='family')])

runtime_catalogue(family)[source]

Load one named runtime catalogue family.

Return type:

object

Parameters:

family (str)

legal_reference(reference_id)[source]

Load one legal declaration by canonical identity.

Return type:

LegalReference

Parameters:

reference_id (str)

source_reference(reference_id)[source]

Load one public-source declaration by canonical identity.

Return type:

SourceReference

Parameters:

reference_id (str)

export_layout(modelo_id, revision_id, layout_id)[source]

Load one separately addressable export layout.

Return type:

ExportLayoutDefinition

Parameters:
legal_evidence(legal_reference_id)[source]

Load one publisher-captured legal evidence projection.

Return type:

PublishedLegalEvidence

Parameters:

legal_reference_id (str)

legal_reference_ids()[source]

Return every published legal declaration identity without hydrating payloads.

Return type:

tuple[str, ...]

legal_quotation_is_grounded(legal_ref_id, quotation)[source]

Answer one citation query from this generation’s published legal evidence.

Return type:

bool

Parameters:
  • legal_ref_id (LegalRefId)

  • quotation (str)

capture_law_selected_projection(modelo_id, *, filing_year, period, on=None, grade=None)[source]

Capture a law-selected inspection or grade-admitted snapshot from this generation.

grade=None selects the static inspection; supplying a grade selects the snapshot admission path. The capture compares current against every coordinate read from the same admitted generation in this process.

Return type:

RegistryAuthorityCapture

Parameters:
read_current_coordinate()[source]

Return the current coordinate captures from this generation compare against.

Return type:

RegistryAuthorityCurrentCoordinate

source_evidence(source_reference_id)[source]

Load one publisher-captured public source payload.

Return type:

PublishedSourceEvidence

Parameters:

source_reference_id (str)

class IndexedRegistryAuthority(descriptor_path)[source]

Bases: object

Own the admitted SQLite reader and issue generation-pinned operation leases.

Parameters:

descriptor_path (Path)

operation()[source]

Pin one reader incarnation for a complete application operation.

Return type:

Generator[PinnedAuthorityOperation]

close()[source]

Close the reader after every operation lease has ended.

Return type:

None

bundled_indexed_authority()[source]

Return the process-shared descriptor-following indexed authority owner.

Return type:

IndexedRegistryAuthority

bundled_authority_descriptor_path()[source]

Return the selector for the content-addressed SQLite generation.

Resolution has two arms and no fallback between them. When cadrumo_authority_root is set it is the whole answer: the descriptor is read from that directory, and its absence there is a refusal rather than a silent slide back to the packaged copy, which would let a development checkout answer from a stale generation it believed it had replaced. When the setting is unset – the installed posture – the packaged resource resolves exactly as it always has.

Return type:

Path

Returns:

The descriptor path, which is guaranteed to exist at the moment of the call.

Raises:

AuthorityDescriptorUnavailableError – When the selected arm carries no descriptor. Fail-closed: every registry read follows this selector, so an absent authority cannot be answered partially.