cadrumo.application.live.notifications module

Bucket-scoped notifications snapshot service.

Wraps the read-only AEAT sede notifications adapter (cadrumo.adapters.outbound.aeat.sede.notifications) with bucket-scoped persistence and a read-only verb surface. The service persists snapshots captured by an upstream fetch, exposes list / show / latest, and never invokes AeatAccessGate.require_live_write().

Submission is permanently forbidden at this boundary: the service has no submit method, no acknowledge method, and no method that calls AEAT to mutate notification state. The acuse (read-receipt) lifecycle is handled locally by tracking which snapshot rows the operator has reviewed.

Verbs:

capture(snapshot) persist a fresh snapshot, emit bucket event latest() return the most recent stored snapshot list_snapshots() return every snapshot in capture order show(snapshot_id) return one snapshot by id

The fetch path itself (HTML parse, auth-gated walker, require_live_read invocation) belongs to the entrypoint that wires the adapter to this service; this module does not import anything that drives a browser.

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 NotificationsSnapshotNotFoundError(message=None, *, context=None, translated_message=None)[source]

Bases: SnapshotNotFoundError

Raised when a snapshot lookup misses by id.

Parameters:

  • message (str | None)

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

  • translated_message (str | None)

Return type:

None

code
class PersistedNotificationsSnapshot(**data)[source]

Bases: BaseModel

A captured snapshot persisted to the active bucket.

snapshot_id is the SHA-256 hex of the canonical JSON form of the underlying NotificationsSnapshot, so two equal snapshots serialise to identical ids and we can deduplicate captures cheaply.

Parameters:

data (Any)

snapshot_id: SnapshotId
bucket_id: BucketId
captured_at: datetime
source_url: str
authenticated_identity: str | None
rows: tuple[RemoteNotification, ...]
persisted_at: datetime
notifications_snapshot_object_key(bucket_id, snapshot_id)[source]

Execute this public contract operation.

Return type:

str

Parameters:

  • bucket_id (str)

  • snapshot_id (str)

class NotificationsService(*, ports)[source]

Bases: StatelessSnapshotService[PersistedNotificationsSnapshot, _NotificationsCaptureRequest]

Bucket-scoped persistence + read surface over notifications snapshots.

The service is structurally read-only. There is no submit verb, no path that could trigger a write to AEAT, and no method that mutates AEAT-side state. The local persistence flow records what was already observed; future fetches re-record state on each capture and emit a fresh bucket event.

Each public verb accepts bucket_id per call; storage is one encrypted secure-object row per captured snapshot.

Parameters:

ports (NotificationsPorts)

capture(*, bucket_id, snapshot, authenticated_identity=None)[source]

Persist a snapshot for the active bucket and return the PersistedNotificationsSnapshot.

The caller is responsible for emitting the corresponding live.notifications.snapshot_captured bucket event; this service does not couple to the event repository so the persistence can be tested in isolation.

Return type:

PersistedNotificationsSnapshot

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

Look up a snapshot by full id or any unambiguous prefix.

Returns the PersistedNotificationsSnapshot that matches snapshot_id within bucket_id.

Return type:

PersistedNotificationsSnapshot

Parameters:

  • bucket_id (str)

  • snapshot_id (str)

latest(*, bucket_id)[source]

Return the most recent PersistedNotificationsSnapshot, or None if none captured.

Return type:

PersistedNotificationsSnapshot | None

Parameters:

bucket_id (str)

async capture_notifications(*, bucket_id, ports, certificate_secret_backend_factory, browser_session_factory, operator_scope_ports)[source]

Capture the authenticated taxpayer’s notifications as encrypted local evidence.

Return type:

PersistedNotificationsSnapshot

Parameters:
resolve_notification_row(*, bucket_id, certificado_id, ports)[source]

Resolve the newest stored notification row for one certificado identifier.

Return type:

RemoteNotification

Parameters:
async pull_notification_document(*, bucket_id, certificado_id, ports, service, certificate_secret_backend_factory, browser_session_factory, operator_scope_ports)[source]

Fetch encrypted custody for a notification that AEAT already records as read.

Parameters: