aeat.application.live._censo module¶

Application-live persistence for captured Modelo 036 censo snapshots.

CensoSnapshot holds the AEAT-side censo facts the operator’s profile must mirror. AEAT is the binding legal source of truth for censo data; the local profile is a cache that must be kept honest. Snapshot records are persisted as Envelope objects through a SecureObjectRepository at IDENTITY SensitivityClass under the censo namespace.

The snapshot pattern mirrors aeat.application.live._borrador_100: content-addressed snapshot ids, encrypted SQLite persistence under a namespaced secure-object key, and a closed ACTIVE / SUPERSEDED / DISCARDED state machine. Re-fetch auto-supersedes the prior ACTIVE snapshot for the same profile.

The CLI-facing CensoSyncService is the only caller; the sede G313 adapter populates censo_facts from the live Mis Datos Censales endpoint.

exception CensoSnapshotNotFoundError(message=None, *, context=None, suggestion=None, translated_message=None)[source]¶

Bases: SnapshotNotFoundError

Raised when a Modelo 036 censo snapshot lookup misses by id.

SnapshotNotFoundError inherits AeatError first, so MRO routes __init__ through the structured constructor. Listing AeatError explicitly here would violate C3 linearization.

Parameters:

message (str | None)
context (Mapping[str, object] | None)
suggestion (str | None)
translated_message (str | None)

Return type:

None

code: ClassVar[ErrorCode]¶

class CensoSnapshot(**data)[source]¶

Bases: BaseModel

Captured 036 censo facts available to application consumers.

censo_facts is a flat mapping keyed by the dotted schema path (e.g. censo.activity_start_date, vivienda_office.office_m2, contact.fiscal_address_cadastral_reference). Values are either decimal strings (for the raw m2 inputs) or short literals (enum values, ISO dates, NIF strings).

Variables:

snapshot_id – Content-addressed SHA-256 hex over (profile_id, captured_at, source_url, censo_facts).
bucket_id – Active profile bucket id at capture time. Snapshots are bucket-scoped so cross-profile leakage is impossible.
profile_id – Operator profile identifier the snapshot belongs to. Carried separately from bucket_id so multi-profile buckets remain addressable.
captured_at – UTC timestamp at which the sede read completed.
source_url – The G313 sede endpoint the snapshot was extracted from. Audited so operators can trace each capture back to its AEAT origin.
state – Lifecycle state from SnapshotLifecycleState.
censo_facts – Flat mapping from dotted schema path to the AEAT-side value. The CensoSyncService.compare verb walks this mapping against the local profile.
superseded_by_snapshot_id – Pointer to the snapshot that replaced this one. Required when state is SUPERSEDED; absent otherwise.
discarded_at – Timestamp captured when the operator explicitly retired the snapshot.
discarded_by – Actor label captured when the operator explicitly retired the snapshot.
discard_reason – Audit reason captured when the operator explicitly retired the snapshot.

Parameters:

snapshot_id (str)
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)])
profile_id (str)
captured_at (datetime)
source_url (str)
state (SnapshotLifecycleState)
censo_facts (Mapping[str, _CensoFactValue])
superseded_by_snapshot_id (str | None)
discarded_at (datetime | None)
discarded_by (str)
discard_reason (str)

snapshot_id: str¶

bucket_id: BucketId¶

profile_id: str¶

captured_at: datetime¶

source_url: str¶

state: SnapshotLifecycleState¶

censo_facts: Mapping[str, _CensoFactValue]¶

superseded_by_snapshot_id: str | None¶

discarded_at: datetime | None¶

discarded_by: str¶

discard_reason: str¶

censo_snapshot_object_key(bucket_id, snapshot_id)[source]¶

Return the secure-object key for one bucket’s censo snapshot.

Return type:

str

Parameters:

bucket_id (str)
snapshot_id (str)

derive_censo_snapshot_id(*, profile_id, captured_at, source_url, censo_facts)[source]¶

Return the content-addressed id for one censo capture.

Two structurally identical snapshots (same profile, same captured- at instant, same source, same fact values) produce the same id; re-saving is then a no-op via CensoSnapshotService.refresh.

Return type:

str

Parameters:

profile_id (str)
captured_at (datetime)
source_url (str)
censo_facts (Mapping[str, _CensoFactValue])

class CensoSnapshotRepository(*, bucket_id, objects=None)[source]¶

Bases: object

Secure-DB repository for captured 036 censo snapshots.

Composes the shared SecureSnapshotRepository (one canonical encrypted secure-object snapshot store) instead of re-implementing the load / resolve / list / save / exists boilerplate. The public class identity, method signatures, CensoSnapshotNotFoundError messages, and captured_at list ordering are preserved; the backing secure-object store is constructed lazily on first use so the repository can be instantiated before a runtime bucket is active.

Parameters:

bucket_id (str)
objects (SecureObjectRepository | None)

property bucket_id: str¶: Return the profile bucket this repository is scoped to.

exists(snapshot_id)[source]¶

Report whether a censo snapshot is persisted under the given id.

Return type:: bool
Parameters:: snapshot_id (str)

load(snapshot_id)[source]¶

Load and return the CensoSnapshot stored under snapshot_id.

Return type:: CensoSnapshot
Parameters:: snapshot_id (str)

list_snapshots()[source]¶

Return every CensoSnapshot for this bucket, oldest capture first.

Return type:: tuple[CensoSnapshot, ...]

resolve(snapshot_id)[source]¶

Resolve an exact or unambiguous-prefix snapshot id to a single CensoSnapshot.

Return type:: CensoSnapshot
Parameters:: snapshot_id (str)

save(snapshot)[source]¶

Persist a censo snapshot into this bucket’s secure-object store.

Return type:: None
Parameters:: snapshot (CensoSnapshot)

class CensoSnapshotService(*, bucket_id, repository=None)[source]¶

Bases: SnapshotService[CensoSnapshot]

Canonical backend service for bucket-scoped 036 censo snapshots.

Mirrors Borrador100SnapshotService. The CLI’s CensoSyncService.refresh_censo is the only caller of capture(); show_censo reads via latest_active() and resolve_snapshot; apply_censo_to_profile reads via latest_active().

The caller is responsible for emitting the CENSO_REFRESHED bucket event after a successful capture (the snapshot service itself is intentionally event-free so the same machinery can be exercised from test scaffolding without polluting the bucket-event-history catalogue).

Parameters:

bucket_id (str)
repository (CensoSnapshotRepository | None)

capture(*, profile_id, captured_at, source_url, censo_facts)[source]¶

Persist a new censo snapshot and return the CensoSnapshot; auto-supersedes the prior ACTIVE.

Re-capturing structurally identical facts (same profile / time / source / values) is a no-op: the existing snapshot is loaded and returned without supersession.

Return type:

CensoSnapshot

Parameters:

profile_id (str)
captured_at (datetime)
source_url (str)
censo_facts (Mapping[str, _CensoFactValue])

list_snapshots(*, profile_id=None, state=SnapshotLifecycleState.ACTIVE)[source]¶

Return CensoSnapshot records for this bucket, narrowed by profile and state.

Widens the base service’s listing with two optional filters. Each snapshot moves through a closed lifecycle of ACTIVE, SUPERSEDED, then DISCARDED states. By default only ACTIVE snapshots are returned, since re-fetching supersedes the prior capture for the same profile.

Parameters:

profile_id (str | None) – When given, keep only snapshots for this operator profile (matched against the trimmed id).
state (SnapshotLifecycleState | None) – Lifecycle state to keep; defaults to SnapshotLifecycleState.ACTIVE. Pass None to keep snapshots in every state.

Returns:

Oldest capture first.

Return type:

tuple[CensoSnapshot, …]

latest_active(*, profile_id)[source]¶

Return the most recently captured ACTIVE snapshot for a profile.

Serves the show_censo and apply_censo_to_profile CLI verbs, which need the single current view of an operator’s Modelo 036 census facts. Selects the ACTIVE snapshot with the latest captured_at for the given profile.

Parameters:: profile_id (str) – Operator profile whose latest capture is wanted.
Returns:: The newest ACTIVE snapshot, or None when the profile has no ACTIVE snapshot.
Return type:: CensoSnapshot

discard(*, snapshot_id, discarded_by, discard_reason='')[source]¶

Mark a snapshot as DISCARDED and return the updated CensoSnapshot. Local-only; never contacts AEAT.

Used when the operator explicitly retires a snapshot captured from a sede outage or with malformed values. The discard does not delete the secure object; it transitions the state so downstream consumers ignore the snapshot.

Return type:

CensoSnapshot

Parameters:

snapshot_id (str)
discarded_by (str)
discard_reason (str)