cadrumo.application.aggregation.observation_window module

Shared set-replace algorithm for per-(modelo, filing_year, period) observation stores.

The percepciones (Modelo 190) and retención (Modelo 180/193) per-perceptor observation repositories each persist a distinct row type keyed by (modelo, filing_year, period) plus a per-record identity, and each needs the exact same SET-REPLACE discipline on a re-pull: clear every prior row for the window, then write the supplied set. Leaving the stale window behind (an additive upsert instead of a set-replace) inflates the next calculate’s distinct count with a record the operator no longer declares — a silent over-count. This module holds that one algorithm so PercepcionObservationRepository and RetencionObservationRepository both call it instead of each carrying their own copy of the loop.

It also holds hashed_tax_id_token(), the one normalise-then-hash step both repositories’ opaque object-key builders need. The two stores stay intentionally distinct (different distinct keys, different models — see each module’s own docstring), but this single identity-hashing primitive is identical between them and belongs here rather than duplicated per file.

hashed_tax_id_token(tax_id, *, field_name)

Return the sha256 of a tax id’s canonical identity token.

Normalises through the same tax_id_identity_token() each observation model applies on construction, so an object key and its aggregation grouping key derive from one identity. The normalisation is idempotent, so an observation-sourced tax id passes through unchanged; the blank guard remains as a defence for callers that key without an observation.

Parameters:

• tax_id (str) – The perceptor’s tax id, in any of the forms tax_id_identity_token() accepts.

• field_name (str) – The field name to report in the blank-token refusal, so each caller’s diagnostic names its own parameter.

Return type:

str

This stays package-private on purpose. The live IVA surface composes the same two steps for its own subject ref and deliberately does not call this. The two callers want OPPOSITE blank behaviour: refusing is required here, where a perceptor with no identifier cannot be keyed, and wrong there, where a subjectless row is a legitimate domain value that must still project. A shared function cannot both raise and not raise, so that caller would guard blank before calling and consume only the hash.

The return width is the second reason and the sharper one: this returns the full digest, while the live ref is truncated, and a truncation width IS an identity contract – two refs compare equal only if every producer cuts them identically. A shared function returning full hex leaves that caller cutting locally anyway.

What is left to share is a single expression over tax_id_identity_token(), which is already the one identity authority both go through.

Note for anyone re-opening this: the domain-neutral refusal a relocation would need already exists – IdentityError with errors.identity.document_empty, shipped in all four catalogues. So “the error would leak retenciones vocabulary” is NOT a reason to keep these apart, and was wrongly given as one when this note was first written. The reasons are the opposite blank contracts and the width.

Raises:

AggregationValidationError – tax_id normalises to a blank token.

Parameters:

• tax_id (str)

• field_name (str)

Return type:

str

class ObservationWindowRepository(*args, **kwargs)

Bases: Protocol, Generic

Required application capability for the set-replace window algorithm.

validate_observation_window_modelo(modelo)

Validate the logical window owner at the outer persistence boundary.

Return type:

str

Parameters:

modelo (str)

extract_identifier(payload)

Return the natural identifier for one persisted payload.

Return type:

str

Parameters:

payload (PayloadT)

iter_records()

Iterate the repository’s decrypted payloads.

Return type:

Iterable[PayloadT]

replace_records(replacements, stale_identifiers)

Atomically replace the selected observation window.

Return type:

None

Parameters:

• replacements (Sequence[PayloadT])

• stale_identifiers (Iterable[str])

replace_observation_window(repository, *, modelo, filing_year, period, observations, source_kind, build_payload, captured_at=None, source_metadata=None)

Replace the FULL observation set for one (modelo, filing_year, period) window.

SET-REPLACE, not additive upsert: every prior row for the exact key-tuple on repository is cleared and the supplied observations are written in its place. An empty observations clears the window, matching each repository’s own no-silent-under-declaration contract on the caller side.

The clear and the write commit as ONE transaction through the repository port’s replace_records operation. Deleting each stale row and then saving each replacement one at a time made every intermediate state observable: a failure part-way through the write loop left the window holding neither the old declared set nor the new one, and the next calculate read that partial window as the operator’s declared truth — a silent under-count with the evidence for it already destroyed.

Parameters:

• repository (ObservationWindowRepository[TypeVar(PayloadT, bound= BaseModel)]) – The window’s encrypted store.

• modelo (str) – Modelo id owning the window.

• filing_year (int) – Filing year owning the window.

• period (Period) – Period owning the window.

• observations (Sequence[TypeVar(ObservationT)]) – The full replacement set (possibly empty).

• source_kind (str) – Capture provenance recorded on every written row.

• build_payload (Callable[..., TypeVar(PayloadT, bound= BaseModel)]) – Each repository’s own typed envelope-payload builder. Returns the payload rather than persisting it, so every row is constructed and validated before any of them is committed.

• captured_at (datetime | None) – Capture instant stamped on every written row; defaults to now, resolved once so the whole set shares one instant.

• source_metadata (Mapping[str, str] | None) – Capture metadata recorded on every written row.

Return type:

None