aeat.application.live._snapshot_base module

Shared lifecycle base classes and helpers for live snapshot services.

This module factors the duplicated state-machine, supersession, and content- addressed-id derivation logic shared across the bucket-scoped live snapshot services (Borrador100, Censo, Expedientes, and Notifications). Each concrete service writes and reads snapshot payloads through a SecureObjectRepository scoped to the active profile bucket.

Design notes:

• SnapshotLifecycleState carries the three operator-visible states all stateful snapshot services share. Every stateful service binds payload state directly to this enum.

• SnapshotRepository is a Protocol — not an abstract class — so concrete per-service repositories (which need to bind a specific TPayload model and a domain-specific object-key prefix) do not need to inherit from it. The service base accepts any object that structurally satisfies the protocol.

• SnapshotService is a generic abstract base whose capture template method coordinates the dedup-by-content-id, auto-supersession, and late-arrival demotion flow. Subclasses implement _payload_axis_key and _build_active_payload to express their domain axis and construction contract; everything else (state transitions, repository orchestration) is shared.

• StatelessSnapshotService is the append-only base for services (Expedientes, Notifications) with no state machine. It accepts bucket_id per call and constructs a fresh repository for the call from an injected repository_factory — the natural shape for services whose public verbs are themselves multi-bucket. Supersession and discard helpers are deliberately absent.

exception SnapshotNotFoundError(message=None, *, context=None, suggestion=None, translated_message=None)

Bases: AeatError, KeyError

Shared base for per-service snapshot-lookup-miss errors.

Inherits from both aeat.core.errors.AeatError and KeyError so the class is enrolled in the ERROR_REGISTRY via the AeatError.__init_subclass__ hook while preserving the mapping-style lookup-miss type. AeatError is listed first so MRO routes __init__ through AeatError.__init__ (which accepts the structured suggestion= / context= kwargs) rather than KeyError’s C-level constructor.

Per-service subclasses (BorradorSnapshotNotFoundError, ExpedientesSnapshotNotFoundError, NotificationsSnapshotNotFoundError, and future siblings) inherit from this base alongside aeat.core.errors.AeatError so callers can either catch the domain-specific class name or the shared parent.

Parameters:

• message (str | None)

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

• suggestion (str | None)

• translated_message (str | None)

Return type:

None

code: ClassVar[ErrorCode]

class SnapshotLifecycleState(*values)

Bases: StrEnum

Lifecycle states shared across stateful live snapshot services.

• ACTIVE — current valid capture; readers consume this.

• SUPERSEDED — replaced by a newer ACTIVE capture on the same axis; retained for audit. Carries superseded_by_snapshot_id.

• DISCARDED — explicitly retired by an operator; carries actor + reason audit metadata.

ACTIVE

SUPERSEDED

DISCARDED

class SnapshotRepository(*args, **kwargs)

Bases: Protocol, Generic

Structural contract for bucket-scoped snapshot persistence backends.

Implementations may be SecureObjectRepository-backed (Borrador100, Censo) or file-system-backed (stateless services).

property bucket_id: str

exists(snapshot_id)

Return type:

bool

Parameters:

snapshot_id (str)

load(snapshot_id)

Return type:

TPayload

Parameters:

snapshot_id (str)

list_snapshots()

Return type:

tuple[TPayload, …]

resolve(snapshot_id)

Return type:

TPayload

Parameters:

snapshot_id (str)

save(snapshot)

Return type:

None

Parameters:

snapshot (TPayload)

derive_snapshot_id_from_json(parts)

Return a SHA-256 content-addressed id for a canonical JSON dict.

parts is serialized with sort_keys=True, ASCII-safe, and the compact (",", ":") separator, then hashed. Callers must pre-coerce Decimals / datetimes / typed-IDs to JSON-safe scalars; the helper does not introspect Pydantic models.

Return type:

str

Parameters:

parts (dict[str, str | int | float | bool | None | list[Any] | dict[str, Any]])

enforce_snapshot_state_invariants(*, state, has_supersession_pointer, discarded_at, discarded_by, discard_reason='')

Enforce the three-state lifecycle invariants for any snapshot payload.

ACTIVE: no supersession pointer, no discard audit metadata. SUPERSEDED: requires supersession pointer, no discard audit metadata. DISCARDED: forbids supersession pointer, requires actor + timestamp.

Domain-specific Pydantic model validators wrap this helper so the same rules apply across Borrador100, Censo, and future stateful services.

Return type:

None

Parameters:

• state (SnapshotLifecycleState)

• has_supersession_pointer (bool)

• discarded_at (datetime | None)

• discarded_by (str)

• discard_reason (str)

class SnapshotService(*, bucket_id, repository)

Bases: ABC, Generic

Abstract lifecycle service base for stateful bucket-scoped snapshots.

Subclasses bind TPayload to their concrete Pydantic snapshot model and implement two hooks:

• _payload_axis_key — returns a tuple identifying the domain axis on which prior ACTIVE snapshots are superseded (e.g. (modelo, year, period) for Borrador100, (profile_id,) for Censo).

• _build_active_payload — constructs an ACTIVE snapshot from keyword-only capture arguments and a derived snapshot id.

The capture template orchestrates dedup, auto-supersession of prior ACTIVE snapshots, and late-arrival demotion when a freshly-captured snapshot arrives older than the current ACTIVE on the same axis.

Parameters:

• bucket_id (str)

• repository (SnapshotRepository[TPayload])

list_snapshots()

Return type:

tuple[TPayload, …]

resolve_snapshot(snapshot_id)

Return type:

TPayload

Parameters:

snapshot_id (str)

class StatelessSnapshotService(*, repository_factory)

Bases: ABC, Generic

Append-only base for stateless snapshot services with per-call buckets.

Subclasses inject a repository_factory that returns a fresh SnapshotRepository for a given bucket id; each public verb accepts bucket_id and materialises the repository on demand. The per-bucket repository is responsible for storage layout and bucket isolation; the base provides the shared dedup, list, and resolve logic.

Subclasses implement two hooks: _derive_snapshot_id and _build_payload. _build_payload receives the resolved bucket_id so payload models can record it on the persisted record.

Parameters:

repository_factory (Callable[[str], SnapshotRepository[TPayload]])

list_snapshots(*, bucket_id)

Return type:

tuple[TPayload, …]

Parameters:

bucket_id (str)

resolve_snapshot(*, bucket_id, snapshot_id)

Return type:

TPayload

Parameters:

• bucket_id (str)

• snapshot_id (str)

class SecureSnapshotRepository(*, bucket_id, payload_model, namespace_definition, object_key, not_found_factory, ambiguous_prefix_factory, domain_label, objects=None)

Bases: Generic

Generic secure-object snapshot repository for one runtime bucket.

The repository preserves the SnapshotRepository structural contract used by stateless live services while replacing one-file-per- bucket JSONL stores with encrypted secure-object rows. Each row is a typed Envelope whose object key carries the bucket id and content-addressed snapshot id.

Parameters:

• bucket_id (str)

• payload_model (type[TPayload])

• namespace_definition (SecureObjectNamespaceDefinition)

• object_key (Callable[[str, str], str])

• not_found_factory (Callable[[str], Exception])

• ambiguous_prefix_factory (Callable[[str, tuple[str, ...]], Exception])

• domain_label (str)

• objects (SecureObjectRepository | None)

property bucket_id: str

exists(snapshot_id)

Return type:

bool

Parameters:

snapshot_id (str)

load(snapshot_id)

Return type:

TPayload

Parameters:

snapshot_id (str)

list_snapshots()

Return type:

tuple[TPayload, …]

resolve(snapshot_id)

Return type:

TPayload

Parameters:

snapshot_id (str)

save(snapshot)

Return type:

None

Parameters:

snapshot (TPayload)