aeat.application.live._notifications module

Bucket-scoped notifications snapshot service.

Wraps the read-only AEAT sede notifications adapter (aeat.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, suggestion=None, translated_message=None)[source]

Bases: SnapshotNotFoundError

Raised when a 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 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:

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

  • rows (tuple[RemoteNotification, ...])

  • persisted_at (datetime)

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]
Return type:

str

Parameters:

  • bucket_id (str)

  • snapshot_id (str)

class NotificationsService(settings=None)[source]

Bases: StatelessSnapshotService[PersistedNotificationsSnapshot]

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:

settings (Settings | None)

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)