aeat.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 a SecureObjectRepository scoped to the active profile bucket.

  • 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. Each observation is wrapped in an Envelope before being written to the secure store.

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, suggestion=None, translated_message=None)[source]

Bases: AeatError

Raised when a verify-observation 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 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.

Parameters:

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

  • surface (VerifySurface)

  • nif (str)

  • verdict (Literal['valid', 'invalid', 'unknown'])

  • expected (Literal['valid', 'invalid', 'unknown'] | None)

  • matched_expectation (bool | None)

  • checked_at (datetime)

  • raw_evidence_locator (str | None)

  • persisted_at (datetime)

observation_id: str
bucket_id: BucketId
surface: VerifySurface
nif: str
verdict: VerifyVerdict
expected: VerifyVerdict | 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 VerifyObservationRepository(*, bucket_id, objects=None)[source]

Bases: object

Secure-object repository for bucket-scoped verify observations.

Parameters:
property bucket_id: str
load(observation_id)[source]

Return the VerifyObservation for observation_id, or None if absent.

Parameters:

observation_id (str) – The full 64-character SHA-256 hex observation id.

Raises:

LiveApplicationInputError – When the loaded observation’s bucket_id or observation_id does not match the repository’s own bucket or the requested id.

Return type:

VerifyObservation | None

list_observations()[source]

Return all stored observations as a tuple of VerifyObservation sorted by check time.

Return type:

tuple[VerifyObservation, ...]

save(observation)[source]

Persist observation as an encrypted Envelope in the object store.

Parameters:

observation (VerifyObservation) – The VerifyObservation to persist. Its bucket_id must match the repository’s own bucket.

Raises:

LiveApplicationInputError – When observation.bucket_id does not match the repository’s bucket id.

Return type:

None

class VerifyService(settings=None)[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:

settings (Settings | None)

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:
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: