cadrumo.adapters.persistence.profile.percepciones_observations module

Encrypted persistence adapter for percepciones observation capabilities.

Core types: SecureObjectRepository, SensitivityClass.

class PercepcionObservationRepositoryAdapter(*, objects)

Bases: SecureBoundRepository[_PercepcionObservationEnvelopePayload], PercepcionObservationRepository

Bind the percepciones port to encrypted secure-object storage.

Parameters:

objects (SecureObjectRepository)

namespace: ClassVar[str]

sensitivity: ClassVar[SensitivityClass]

schema_version: ClassVar[int]

payload_type

alias of _PercepcionObservationEnvelopePayload

validate_observation_window_modelo(modelo)

Bind application window-key validation to persistence storage safety.

Return type:

str

Parameters:

modelo (str)

extract_identifier(payload)

Return the natural id for payload (used as the SQL object key).

Subclasses MUST override. The base implementation raises NotImplementedError.

Return type:

str

Parameters:

payload (_PercepcionObservationEnvelopePayload)

build_observation_payload(*, modelo, filing_year, period, observation, source_kind, captured_at=None, source_metadata=None)

Build one validated envelope payload before committing it.

Return type:

_PercepcionObservationEnvelopePayload

Parameters:

• modelo (str)

• filing_year (int)

• period (Period)

• observation (WithholdingObservation)

• source_kind (AggregationCaptureKind)

• captured_at (datetime | None)

• source_metadata (Mapping[str, str] | None)

save_observation(*, modelo, filing_year, period, observation, source_kind, captured_at=None, source_metadata=None)

Persist one per-perceptor-clave observation.

Return type:

None

Parameters:

• modelo (str)

• filing_year (int)

• period (Period)

• observation (WithholdingObservation)

• source_kind (AggregationCaptureKind)

• captured_at (datetime | None)

• source_metadata (Mapping[str, str] | None)

replace_observations(*, modelo, filing_year, period, observations, source_kind, captured_at=None, source_metadata=None)

Atomically replace the complete per-perceptor-clave 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)

Return all persisted observations for one modelo/period window.

Return type:

tuple[WithholdingObservation, ...]

Parameters:

• modelo (str)

• period (Period)

iter_modelo(modelo)

Yield encrypted per-perceptor-clave payloads for one modelo.

Return type:

Iterator[_PercepcionObservationEnvelopePayload]

Parameters:

modelo (str)