aeat.application.live._expedientes module

Bucket-scoped expedientes snapshot service.

Wraps the read-only AEAT sede declarations walker (aeat.adapters.outbound.aeat.sede._declarations) 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, suggestion=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)

  • suggestion (str | None)

  • translated_message (str | None)

Return type:

None

code: ClassVar[ErrorCode]
class ExpedientesCapture(**data)[source]

Bases: BaseModel

Slim wrapper around a Declaracion walker result.

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

Parameters:
declarations: tuple[Declaracion, ...]
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:

  • snapshot_id (Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=64, max_length=64, pattern=^[0-9a-f]{64}$, ascii_only=None)])

  • bucket_id (Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=128, pattern=None, ascii_only=None)])

  • captured_at (datetime)

  • source_url (str)

  • authenticated_identity (str | None)

  • declarations (tuple[Declaracion, ...])

  • persisted_at (datetime)

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

str

Parameters:

  • bucket_id (str)

  • snapshot_id (str)

class ExpedientesService(settings=None)[source]

Bases: StatelessSnapshotService[PersistedExpedientesSnapshot]

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:

settings (Settings | None)

capture(*, bucket_id, capture)[source]
Return type:

PersistedExpedientesSnapshot

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

PersistedExpedientesSnapshot

Parameters:

  • bucket_id (str)

  • snapshot_id (str)

latest(*, bucket_id)[source]
Return type:

PersistedExpedientesSnapshot | None

Parameters:

bucket_id (str)