cadrumo.application.live.expedientes module

Bucket-scoped expedientes snapshot service.

Coordinates the read-only declaration-register capability supplied by the outer composition root with bucket-scoped persistence. Read-only by construction: no method calls AEAT to mutate expediente state.

Verbs:

capture(snapshot) persist a fresh expedientes capture, deduplicated list_snapshots() every captured snapshot, in capture order show(snapshot_id) single snapshot by full id or unambiguous prefix latest() most recent snapshot, or None

The fetch path (auth-gated walker, require_live_read invocation) lives in the entrypoint that wires the adapter to this service.

The lifecycle helpers (content-addressed id derivation, dedup on re-capture, list/show/latest) are routed through the shared StatelessSnapshotService base; the public class identity, exception class names, secure-object storage layout, and per-call bucket_id signatures are preserved exactly.

exception ExpedientesSnapshotNotFoundError(message=None, *, context=None, translated_message=None)[source]

Bases: SnapshotNotFoundError

Raised when an expedientes snapshot lookup misses by id.

Parameters:

  • message (str | None)

  • context (Mapping[str, object] | None)

  • translated_message (str | None)

Return type:

None

code
class ExpedientesCapture(**data)[source]

Bases: BaseModel

Slim wrapper around translated declaration-register rows.

Mirrors the read-only marker pattern from NotificationsSnapshot: mode='read' is the structural assertion that this capture cannot drive an AEAT-side mutation.

Parameters:

data (Any)

declarations: tuple[ExpedientesDeclaration, ...]
captured_at: datetime
source_url: str
authenticated_identity: str | None
mode: str
class PersistedExpedientesSnapshot(**data)[source]

Bases: BaseModel

Captured expedientes snapshot persisted to the active bucket.

Parameters:

data (Any)

snapshot_id: SnapshotId
bucket_id: BucketId
captured_at: datetime
source_url: str
authenticated_identity: str | None
declarations: tuple[ExpedientesDeclaration, ...]
persisted_at: datetime
expedientes_snapshot_object_key(bucket_id, snapshot_id)[source]

Execute this public contract operation.

Return type:

str

Parameters:

  • bucket_id (str)

  • snapshot_id (str)

class ExpedientesService(*, ports)[source]

Bases: StatelessSnapshotService[PersistedExpedientesSnapshot, ExpedientesCapture]

Bucket-scoped persistence + read surface over expedientes snapshots.

Structurally read-only per the live-AEAT charter. No submit, no acknowledge, no method that mutates AEAT state. Each public verb accepts bucket_id per call; storage is one encrypted secure-object row per captured snapshot.

Parameters:

ports (ExpedientesPorts)

capture(*, bucket_id, capture)[source]

Execute this public contract operation.

Return type:

PersistedExpedientesSnapshot

Parameters:
show(*, bucket_id, snapshot_id)[source]

Execute this public contract operation.

Return type:

PersistedExpedientesSnapshot

Parameters:

  • bucket_id (str)

  • snapshot_id (str)

latest(*, bucket_id)[source]

Execute this public contract operation.

Return type:

PersistedExpedientesSnapshot | None

Parameters:

bucket_id (str)

async capture_expedientes(*, bucket_id, modelo, year, ports, certificate_secret_backend_factory, browser_session_factory, operator_scope_ports)[source]

Capture the selected declaration-register view as encrypted local evidence.

Return type:

PersistedExpedientesSnapshot

Parameters:
async capture_expedientes_bulk(*, bucket_id, year_from, year_to, modelos=None, ports, certificate_secret_backend_factory, browser_session_factory, operator_scope_ports, operation)[source]

Capture each requested declaration-register view while reporting isolated failures.

Return type:

ExpedientesBulkCaptureReport

Parameters: