cadrumo.adapters.persistence.profile.retencion_observations module

Encrypted persistence adapter for retención observation capabilities.

Core types: SecureObjectRepository, SensitivityClass.

class RetencionObservationRepositoryAdapter(*, objects)

Bases: SecureBoundRepository[_RetencionObservationEnvelopePayload], RetencionObservationRepository

Bind the application retención port to encrypted secure-object storage.

Parameters:

objects (SecureObjectRepository)

namespace: ClassVar[str]

sensitivity: ClassVar[SensitivityClass]

schema_version: ClassVar[int]

payload_type

alias of _RetencionObservationEnvelopePayload

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 (_RetencionObservationEnvelopePayload)

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:

_RetencionObservationEnvelopePayload

Parameters:

• modelo (str)

• filing_year (int)

• period (Period)

• observation (RetencionObservation)

• 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 retención observation.

Return type:

None

Parameters:

• modelo (str)

• filing_year (int)

• period (Period)

• observation (RetencionObservation)

• 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 observation window.

Return type:

None

Parameters:

• modelo (str)

• filing_year (int)

• period (Period)

• observations (Sequence[RetencionObservation])

• 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[RetencionObservation, ...]

Parameters:

• modelo (str)

• period (Period)

iter_modelo(modelo)

Yield encrypted payloads for one modelo in unspecified order.

Return type:

Iterator[_RetencionObservationEnvelopePayload]

Parameters:

modelo (str)