cadrumo.adapters.outbound.aeat.sede.filed_observation_persistence module

Concrete adapters for the filed-observation application ports.

The live filed-history service owns only protocols. This module is the outer binding for those protocols: it joins the existing Sede parser and observation store to the encrypted calculation, filing, IVA-history, event, and baseline repositories and the filing-chain reconciliation. Adapter failures are translated at this boundary so an application caller never has to know a Sede or secure-object exception type.

Core types: ModeloRecord, SecureObjectRepository.

class FiledObservationParserAdapter(*args, **kwargs)

Bases: FiledObservationParserPort

Adapt receipt parsing and Sede CSV extraction to application types.

parse_justificante(body)

Parse receipt bytes without creating a plaintext temporary file.

Return type:

Justificante

Parameters:

body (bytes)

csv_from_source_url(source_url)

Extract the independent AEAT CSV witness from a source URL.

Return type:

str

Parameters:

source_url (str)

class FiledDeclarationTransformationAdapter(*, operation)

Bases: FiledDeclarationTransformationPort

Adapt Sede observation projections to registry-grounded application ports.

Parameters:

operation (PinnedAuthorityOperation)

registry_observation(observation)

Build the registry-grounded numeric projection.

Return type:

RegistryModeloObservation

Parameters:

observation (FiledObservationProtocol)

non_numeric_casillas(observation)

Return non-numeric observed casillas for the operator projection.

Return type:

Sequence[FiledObservationSkipProtocol]

Parameters:

observation (FiledObservationProtocol)

class FiledObservationStoreAdapter(*, root, objects)

Bases: FiledObservationPersistencePort

Adapt the encrypted Sede observation store to the application port.

Parameters:

• root (Path)

• objects (SecureObjectRepository)

persist_observation(observation)

Persist one encrypted observation manifest.

Return type:

Path

Parameters:

observation (FiledObservationProtocol)

persist_artefact(observation_key, artefact, body)

Persist one encrypted artefact and return its storage reference.

Return type:

FiledObservationArtefactProtocol

Parameters:

• observation_key (tuple[str, int, Period, str])

• artefact (FiledObservationArtefactProtocol)

• body (bytes)

load_artefact(storage_ref)

Load and verify one encrypted artefact by its content address.

Return type:

bytes

Parameters:

storage_ref (str)

class CalculationObservationRepositoryAdapter(*, repository)

Bases: FiledCalculationObservationRepositoryPort

Adapt the encrypted calculation-observation repository.

Parameters:

repository (CalculationObservationRepository)

load_observation(modelo, period)

Load one existing calculation-observation envelope.

Return type:

ObservationEnvelopePayload | None

Parameters:

• modelo (str)

• period (Period)

prepare_observation_envelope(observation, *, source_kind, stamped_revision_id, captured_at=None, source_metadata=None, source_headers=())

Prepare one validated calculation-observation envelope.

Return type:

ObservationEnvelopePayload

Parameters:

• observation (RegistryModeloObservation)

• source_kind (ObservationSourceKind | str)

• stamped_revision_id (str)

• captured_at (datetime | None)

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

• source_headers (tuple[ObservedHeaderFact, ...])

save(payload)

Persist one prepared calculation-observation envelope.

Return type:

None

Parameters:

payload (ObservationEnvelopePayload)

class IvaHistoryRepositoryAdapter(*, repository)

Bases: FiledIvaHistoryRepositoryPort

Adapt the encrypted IVA compensation-history repository.

Parameters:

repository (IvaCompensationHistoryRepository)

load_period(period)

Reload one persisted IVA history period.

Return type:

IvaCompensationPeriodState | None

Parameters:

period (Period)

class IvaObservationPersistenceAdapter(*args, **kwargs)

Bases: FiledIvaObservationPersistencePort

Adapt the canonical atomic Modelo 303/history co-commit operation.

persist(*, observation_repository, history_repository, envelope, taxpayer_nif, source_observation_key, expediente_id, status, source_artefact_sha256)

Atomically persist the calculation envelope and IVA history state.

Return type:

IvaCompensationPeriodState

Parameters:

• observation_repository (FiledCalculationObservationRepositoryPort)

• history_repository (FiledIvaHistoryRepositoryPort)

• envelope (ObservationEnvelopePayload)

• taxpayer_nif (str)

• source_observation_key (str)

• expediente_id (str | None)

• status (str | None)

• source_artefact_sha256 (str | None)

class JustificanteRepositoryAdapter(*, repository)

Bases: JustificanteRepositoryProtocol

Adapt encrypted justificante metadata persistence.

Parameters:

repository (JustificanteRepository)

load(csv, /)

Load one receipt by CSV.

Return type:

Justificante | None

Parameters:

csv (str)

save(justificante, /)

Persist one parsed receipt.

Return type:

None

Parameters:

justificante (Justificante)

iter_justificantes()

Yield all receipts after translating scan failures.

Return type:

Iterator[Justificante]

class FilingRepositoryAdapter(*, repository)

Bases: ModeloRecordCatalogueRepositoryProtocol

Adapt encrypted modelo filing-record catalogue persistence.

Parameters:

repository (ModeloRecordCatalogueRepository)

property bucket_id: str | None

Return the repository’s bucket binding.

exists()

Return whether the filing catalogue exists.

Return type:

bool

load()

Load the filing catalogue.

Return type:

ModeloRecordCatalogue

load_revisioned()

Load the filing catalogue and its revision marker.

Return type:

tuple[ModeloRecordCatalogue, str]

save(catalogue)

Persist the filing catalogue.

Return type:

None

Parameters:

catalogue (ModeloRecordCatalogue)

mutate(mutation)

Apply one guarded filing-catalogue mutation.

Return type:

ModeloRecordCatalogue

Parameters:

mutation (Callable[[ModeloRecordCatalogue], ModeloRecordCatalogue])

to_secure_object_write(catalogue, *, expected_revision_id=None)

Prepare a filing-catalogue secure-object write.

Return type:

SecureObjectWrite

Parameters:

• catalogue (ModeloRecordCatalogue)

• expected_revision_id (str | None)

save_with_secure_object_writes(catalogue, extra_writes, *, expected_revision_id=None)

Persist filing state and co-emitted writes atomically.

Return type:

None

Parameters:

• catalogue (ModeloRecordCatalogue)

• extra_writes (tuple[SecureObjectWrite, ...])

• expected_revision_id (str | None)

class BucketEventRepositoryAdapter(*, repository)

Bases: BucketEventHistoryRepositoryProtocol

Adapt encrypted bucket-event-history persistence.

Parameters:

repository (BucketEventHistoryRepository)

exists()

Return whether the event catalogue exists.

Return type:

bool

load()

Load the event catalogue.

Return type:

BucketEventHistoryCatalogue

save(catalogue)

Persist the event catalogue.

Return type:

None

Parameters:

catalogue (BucketEventHistoryCatalogue)

to_secure_object_write(catalogue, *, expected_revision_id=None)

Prepare an event-catalogue secure-object write.

Return type:

SecureObjectWrite

Parameters:

• catalogue (BucketEventHistoryCatalogue)

• expected_revision_id (str | None)

class FilingReconciliationAdapter(*, work_lifecycle_ports, calculation_repository, filing_repository, justificante_repository, observation_repository)

Bases: FiledFilingReconciliationPort

Adapt the filing-chain reconciliation service with explicit repositories.

Parameters:

• work_lifecycle_ports (WorkLifecyclePorts)

• calculation_repository (CalculationRevisionCatalogueRepository)

• filing_repository (ModeloRecordCatalogueRepository)

• justificante_repository (JustificanteRepository)

• observation_repository (CalculationObservationRepository)

reconcile(entry, *, actor, clock)

Reconcile one AEAT register entry under the bundled authority.

Return type:

FilingReconciliationResult

Parameters:

• entry (AeatRegisterEntry)

• actor (str)

• clock (datetime)