cadrumo.application.aggregation.percepciones_observations_repository module¶

Application-owned capabilities for Modelo 190 percepciones observations.

The percepciones source stores one WithholdingObservation for every (perceptor, clave, subclave) in a (modelo, filing_year, period) window. The application owns the key grammar and the required read/write capability; the encrypted persistence implementation is bound by an outer composition root. Keeping this module free of storage imports makes the calculate mesh usable with an inward fake and prevents persistence details from becoming an implicit application dependency.

percepcion_observation_key(modelo, filing_year, period, perceptor_tax_id, clave, subclave)[source]¶

Build the opaque per-perceptor-clave observation key.

The NIF is hashed because object keys are storage metadata. clave and subclave remain explicit because they are non-identifying AEAT codes and distinguish the separate registro-tipo-2 observations.

Return type:

str

Parameters:

modelo (str)
filing_year (int)
period (Period)
perceptor_tax_id (str)
clave (str)
subclave (str)

exception PercepcionObservationPersistenceError(operation)[source]¶

Bases: CadrumoError

Translated failure from the perceptor-observation persistence port.

Parameters:: operation (str)
Return type:: None

code¶

class PercepcionObservationRepository(*args, **kwargs)[source]¶

Bases: Protocol

Application-facing read/write capability for percepciones windows.

replace_observations(*, modelo, filing_year, period, observations, source_kind, captured_at=None, source_metadata=None)[source]¶

Atomically replace the complete observation window.

Return type:

None

Parameters:

modelo (str)
filing_year (int)
period (Period)
observations (Sequence[WithholdingObservation])
source_kind (AggregationCaptureKind)
captured_at (datetime | None)
source_metadata (Mapping[str, str] | None)

load_observations(modelo, period)[source]¶

Return observations for one modelo and filing period.

Return type:

tuple[WithholdingObservation, ...]

Parameters:

modelo (str)
period (Period)

class PercepcionObservationPorts(repository)[source]¶

Bases: object

Required percepciones capabilities for one profile calculation context.

Parameters:: repository (PercepcionObservationRepository)

repository: PercepcionObservationRepository¶

class PercepcionObservationPortsFactory(*args, **kwargs)[source]¶

Bases: Protocol

Construct the percepciones capabilities for one profile bucket.

persist_percepcion_observations(*, ports, modelo, filing_year, period, observations, source_kind=AggregationCaptureKind.AGGREGATE_PULL)[source]¶

Persist the complete per-perceptor-clave window through the required port.

Return type:

None

Parameters:

ports (PercepcionObservationPorts)
modelo (str)
filing_year (int)
period (Period)
observations (Sequence[WithholdingObservation])
source_kind (AggregationCaptureKind)