cadrumo.application.live.verify module

Bucket-scoped verify service.

Wraps the two read-only AEAT verify oracles into a bucket-scoped audit log. Verify observations are persisted through an application-owned observation persistence port supplied by the outer composition.

  • NIF-IVA (VIES) — intracomunitario counterparty validation

  • TGVI / GROI — intra-community operator (registered Spanish NIF)

Both surfaces are on-demand single-shot checks. The service records each check as a typed observation tied to the active bucket so the operator can audit which NIFs were verified, when, and against what verdict. Subsequent invocations against the same NIF produce a new observation row; history is never overwritten. The persistence implementation decides how each observation is protected and addressed at rest; this use case consumes only its application-level contract.

Structurally read-only:

  • the service has no submit / mutate verb;

  • the underlying drivers call AeatAccessGate.require_live_read() before remote contact; this layer consumes their results only;

  • the operator’s --expected valid|invalid|unknown hint is recorded alongside the verdict so a mistaken expectation is auditable.

class VerifySurface(*values)[source]

Bases: StrEnum

Closed catalogue of supported verify surfaces.

NIF_IVA
TGVI
exception VerifyObservationNotFoundError(message=None, *, context=None, translated_message=None)[source]

Bases: CadrumoError

Raised when a verify-observation lookup misses by id.

Parameters:

  • message (str | None)

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

  • translated_message (str | None)

Return type:

None

code
class VerifyObservation(**data)[source]

Bases: BaseModel

One persisted verify check.

The observation_id is content-addressed (SHA-256 of canonical fields) so two identical checks against the same NIF on the same timestamp deduplicate without separate id management. It is typed as ContentDigest so the persisted identity carries the canonical lowercase hex-64 digest shape rather than any 64-character string: a malformed id would otherwise reach the secure-object key, the load round-trip, and the show projection unchallenged.

Both instants are UTC-aware. checked_at feeds _derive_observation_id() through isoformat(), so an ambiguous offset would silently fork the content address of the same check; persisted_at orders encrypted history.

Parameters:

data (Any)

observation_id: ContentDigest
bucket_id: BucketId
surface: VerifySurface
nif: str
verdict: IdentityCheckVerdictValue
expected: IdentityCheckVerdictValue | None
matched_expectation: bool | None
checked_at: datetime
raw_evidence_locator: str | None
persisted_at: datetime
verify_observation_object_key(bucket_id, observation_id)[source]

Return the canonical secure-object key for a VerifyObservation.

The key encodes both the bucket and the observation so the store remains globally unique across buckets even when two buckets check the same NIF at the same instant.

Parameters:

  • bucket_id (str) – The profile bucket’s UUIDv4 identifier.

  • observation_id (str) – The SHA-256 hex content-address of the observation.

Raises:

LiveApplicationInputError – When either argument is blank after stripping whitespace.

Return type:

str

class VerifyService(*, persistence)[source]

Bases: object

Bucket-scoped audit log of NIF verify checks.

Structurally read-only. The service has no submit, no mutate, and no method that would alter AEAT-side state. Verify surfaces themselves are read-only by construction; this layer only records observations the drivers produce.

Parameters:

persistence (VerifyObservationPersistencePort)

record(*, bucket_id, surface, nif, verdict, checked_at, expected=None, raw_evidence_locator=None)[source]

Persist one verify observation. Deduplicates identical replays.

Returns a VerifyObservation with the persisted observation id and all supplied fields.

Return type:

VerifyObservation

Parameters:

  • bucket_id (str)

  • surface (VerifySurface)

  • nif (str)

  • verdict (Literal[IdentityCheckVerdict.VALID, IdentityCheckVerdict.INVALID, IdentityCheckVerdict.UNKNOWN])

  • checked_at (datetime)

  • expected (Literal[IdentityCheckVerdict.VALID, IdentityCheckVerdict.INVALID, IdentityCheckVerdict.UNKNOWN] | None)

  • raw_evidence_locator (str | None)

list_observations(*, bucket_id, surface=None, nif=None)[source]

Return all VerifyObservation records in capture order. Optional filters.

Return type:

tuple[VerifyObservation, ...]

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

Look up and return the VerifyObservation for the given full id or unambiguous prefix.

Return type:

VerifyObservation

Parameters:

  • bucket_id (str)

  • observation_id (str)

latest_for_nif(*, bucket_id, surface, nif)[source]

Return the most recent VerifyObservation for (surface, nif), or None.

Return type:

VerifyObservation | None

Parameters: