cadrumo.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 SnapshotRepository scoped to the active profile bucket.

This module declares the PORT and the lifecycle bases only. The encrypted secure-object backend that satisfies the port lives in the persistence adapter, as adapters.persistence.profile.snapshots.SecureSnapshotRepository, so the storage coupling is an adapter-internal edge rather than an application-layer reach outward.

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, translated_message=None)

Bases: CadrumoError

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

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

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

Parameters:

• message (str | None)

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

• translated_message (str | None)

Return type:

None

code

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

SnapshotLifecycleStateValue

The same lifecycle where a CLI row or payload must accept the plain token.

alias of Literal[ACTIVE, SUPERSEDED, DISCARDED]

class SnapshotStateFilter(*values)

Bases: StrEnum

Operator-facing filter over SnapshotLifecycleState, plus ALL.

ALL is deliberately NOT a member of SnapshotLifecycleState. No persisted snapshot is ever in an “all” state, so adding it there would give every exhaustive match over the lifecycle a branch that cannot occur and would let a stored record claim a state that means “no filter”. The filter is its own closed axis that maps onto the lifecycle, mirroring ReviewState.

Every lifecycle state has a member here, so a filter exists for each; the correspondence is enforced by a gate rather than left to the next author.

ACTIVE

SUPERSEDED

DISCARDED

ALL

as_lifecycle_state()

Return the lifecycle state to filter on, or None for no filter.

Return type:

SnapshotLifecycleState | None

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

Execute this public contract operation.

exists(snapshot_id)

Execute this public contract operation.

Return type:

bool

Parameters:

snapshot_id (str)

load(snapshot_id)

Execute this public contract operation.

Return type:

TPayload

Parameters:

snapshot_id (str)

list_snapshots()

Execute this public contract operation.

Return type:

tuple[TPayload, …]

resolve(snapshot_id)

Execute this public contract operation.

Return type:

TPayload

Parameters:

snapshot_id (str)

save(snapshot)

Execute this public contract operation.

Return type:

None

Parameters:

snapshot (TPayload)

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

Execute this public contract operation.

Return type:

tuple[TPayload, …]

resolve_snapshot(snapshot_id)

Execute this public contract operation.

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)

Execute this public contract operation.

Return type:

tuple[TPayload, …]

Parameters:

bucket_id (str)

resolve_snapshot(*, bucket_id, snapshot_id)

Execute this public contract operation.

Return type:

TPayload

Parameters:

• bucket_id (str)

• snapshot_id (str)