aeat.application.ledger._evidence module

Purchase invoice evidence records and the CRUD application service.

aeat app ledger evidence {add|remove|update|view|list} operate over a PurchaseInvoiceEvidence pydantic record. Audit events are emitted to a BucketEventHistoryRepository on every mutating verb.

File-type scope is restricted to PDF and image inputs. Plaintext, email body, and Drive-URL evidence sources are out of scope. add refuses non-PDF/non-image source paths with a typed PurchaseInvoiceEvidenceInputError.

Persistence is bucket-scoped encrypted secure-object storage. The evidence catalogue is a PurchaseInvoiceEvidenceDocument persisted through SecureBoundRepository under aeat.adapters.persistence.storage.LEDGER_PURCHASE_INVOICE_EVIDENCE_NAMESPACE. At add time the source file’s bytes are copied into the encrypted AttachmentStore (active bucket) and the resulting content-addressed attachment_id is recorded on the evidence record; the bytes thereafter live only in secure storage. source_path is retained as a provenance breadcrumb and is never read for bytes (sensitive-financial-data-secure-storage-only).

class MediaKind(*values)[source]

Bases: StrEnum

Canonical media-kind values for purchase invoice evidence.

PDF
IMAGE
exception PurchaseInvoiceEvidenceInputError(message=None, *, context=None, suggestion=None, translated_message=None)[source]

Bases: AeatError

Raised when a CLI-supplied evidence input violates the typed contract.

Parameters:

  • message (str | None)

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

  • suggestion (str | None)

  • translated_message (str | None)

Return type:

None

code: ClassVar[ErrorCode]
exception PurchaseInvoiceEvidenceNotFoundError(message=None, *, context=None, suggestion=None, translated_message=None)[source]

Bases: AeatError

Raised when a CLI lookup targets a missing evidence record.

Parameters:

  • message (str | None)

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

  • suggestion (str | None)

  • translated_message (str | None)

Return type:

None

code: ClassVar[ErrorCode]
class PurchaseInvoiceEvidence(**data)[source]

Bases: BaseModel

One persisted purchase invoice evidence record.

Parameters:

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

  • source_path (str)

  • source_sha256 (str)

  • attachment_id (str | None)

  • media_kind (MediaKind)

  • supplier (str | None)

  • invoice_number (str | None)

  • invoice_date (str | None)

  • taxable_base (Decimal | None)

  • iva_rate (Decimal | None)

  • iva_amount (Decimal | None)

  • notes (str)

  • created_at (datetime)

  • updated_at (datetime)

evidence_id: str
bucket_id: BucketId
source_path: str
source_sha256: str
attachment_id: str | None
media_kind: MediaKind
supplier: str | None
invoice_number: str | None
invoice_date: str | None
taxable_base: Decimal | None
iva_rate: Decimal | None
iva_amount: Decimal | None
notes: str
created_at: datetime
updated_at: datetime
derive_purchase_invoice_evidence_id(*, bucket_id, source_sha256, media_kind, supplier, invoice_number, invoice_date, taxable_base, iva_rate, iva_amount, notes, created_at, disambiguator=0)[source]

Return the content-addressed id for a purchase-invoice evidence record.

Mirrors aeat.domain.transactions.derive_transaction_id(): the id is a SHA-256 digest (truncated to 16 hex chars, the prior surrogate’s width) over the record’s identifying fields, so it is stable under a frozen-clock replay and directly referenceable as an aeat app ledger evidence argument, needing no output mask. created_at plus the disambiguator ordinal preserve the genuine-duplicate case the ledger already supports: two evidence records for the same file must keep distinct ids, so the mint site increments disambiguator on the rare digest collision (identical fields at an identical coarse-clock instant) rather than colliding.

Return type:

str

Parameters:
class PurchaseInvoiceEvidenceDocument(**data)[source]

Bases: BaseModel

Encrypted bucket-local purchase invoice evidence catalogue.

Parameters:

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

  • records (tuple[PurchaseInvoiceEvidence, ...])

bucket_id: BucketId
records: tuple[PurchaseInvoiceEvidence, ...]
class PurchaseInvoiceEvidencePatch(**data)[source]

Bases: BaseModel

Mutable subset of PurchaseInvoiceEvidence fields accepted by update.

Only the fields listed here may be changed after an evidence record is created. evidence_id, bucket_id, source_path, source_sha256, media_kind, and the timestamp fields are immutable. A None value for any optional field means “leave unchanged”; the service ignores None entries when applying the patch.

Parameters:

  • supplier (str | None)

  • invoice_number (str | None)

  • invoice_date (str | None)

  • taxable_base (Decimal | None)

  • iva_rate (Decimal | None)

  • iva_amount (Decimal | None)

  • notes (str | None)

supplier: str | None
invoice_number: str | None
invoice_date: str | None
taxable_base: Decimal | None
iva_rate: Decimal | None
iva_amount: Decimal | None
notes: str | None
class PurchaseInvoiceEvidenceResult(**data)[source]

Bases: BaseModel

Return record from a mutating evidence verb — record plus emitted event id.

Parameters:
record: PurchaseInvoiceEvidence
bucket_event_ids: tuple[str, ...]
class PurchaseInvoiceEvidenceRepository(*, bucket_id=None, objects=None, settings=None)[source]

Bases: SecureBoundRepository[PurchaseInvoiceEvidenceDocument]

Encrypted store for one bucket’s PurchaseInvoiceEvidenceDocument.

The namespace, sensitivity, schema version, and object-key contract come from aeat.adapters.persistence.storage.LEDGER_PURCHASE_INVOICE_EVIDENCE_NAMESPACE. The SecureBoundRepository base wraps each PurchaseInvoiceEvidenceDocument in a Envelope before writing it.

See also

PurchaseInvoiceEvidenceService

CRUD service that mutates this repository and emits bucket events.

AttachmentStore

Encrypted byte store that holds the referenced source files.

Parameters:
namespace: ClassVar[str]
sensitivity: ClassVar[SensitivityClass]
schema_version: ClassVar[int]
payload_type

alias of PurchaseInvoiceEvidenceDocument

extract_identifier(payload)[source]

Return the natural id for payload (used as the SQL object key).

Subclasses MUST override. The base implementation raises NotImplementedError.

Return type:

str

Parameters:

payload (PurchaseInvoiceEvidenceDocument)

class PurchaseInvoiceEvidenceService(settings=None, bucket_event_repository=None)[source]

Bases: object

Application service for the aeat app ledger evidence verb group.

Parameters:
add(*, bucket_id, source_path, supplier=None, invoice_number=None, invoice_date=None, taxable_base=None, iva_rate=None, iva_amount=None, notes='', actor='cli')[source]

Attach a new purchase invoice evidence file to a bucket (ledger).

Resolves source_path to an absolute path, verifies the file exists, infers the MediaKind from the extension, copies the file’s bytes into the encrypted AttachmentStore (active bucket) and records the resulting content-addressed attachment_id on the record (the bytes thereafter live only in secure storage; source_path is a provenance breadcrumb), creates a PurchaseInvoiceEvidence record, appends it to the in-memory catalogue, persists the encrypted bucket-local catalogue in secure-object storage, and emits a PURCHASE_INVOICE_EVIDENCE_ATTACHED audit event.

Parameters:

  • bucket_id (str) – Ledger bucket the evidence belongs to.

  • source_path (Path) – Local path to a PDF or image file.

  • supplier (str | None) – Optional vendor name extracted from the invoice.

  • invoice_number (str | None) – Optional invoice identifier from the document.

  • invoice_date (str | None) – Optional issue date string (free-form; typically YYYY-MM-DD).

  • taxable_base (Decimal | None) – Optional net taxable amount (~decimal.Decimal).

  • iva_rate (Decimal | None) – Optional IVA percentage as a ~decimal.Decimal.

  • iva_amount (Decimal | None) – Optional IVA amount as a ~decimal.Decimal.

  • notes (str) – Operator free-text annotation.

  • actor (str) – Identifier stamped on the audit event (defaults to "cli").

Returns:

Carrying the new record and the emitted audit event id.

Return type:

PurchaseInvoiceEvidenceResult

Raises:

PurchaseInvoiceEvidenceInputError – if source_path is not a readable file or has an unsupported extension.

view(*, bucket_id, evidence_id)[source]

Return the single evidence record identified by evidence_id.

Parameters:

  • bucket_id (str) – Ledger bucket to search.

  • evidence_id (str) – Unique evidence id assigned at add time.

Returns:

The matching record.

Return type:

PurchaseInvoiceEvidence

Raises:

PurchaseInvoiceEvidenceNotFoundError – if no record with that id exists in the bucket.

list_all(*, bucket_id)[source]

Return all evidence records for a bucket in append order.

Parameters:

bucket_id (str) – Ledger bucket to read.

Returns:

Oldest first. Returns an empty tuple if the bucket has no evidence file yet.

Return type:

tuple[PurchaseInvoiceEvidence, …]

update(*, bucket_id, evidence_id, patch, actor='cli')[source]

Apply a partial update to an existing evidence record.

Loads the bucket’s record list, finds the record matching evidence_id, merges non-None fields from patch, stamps updated_at, writes the updated list back, and emits a PURCHASE_INVOICE_EVIDENCE_REPLACED audit event.

Parameters:

  • bucket_id (str) – Ledger bucket containing the record.

  • evidence_id (str) – Id of the record to update.

  • patch (PurchaseInvoiceEvidencePatch) – PurchaseInvoiceEvidencePatch carrying the fields to change. Fields set to None are left unchanged.

  • actor (str) – Identifier stamped on the audit event.

Returns:

With the updated record and audit event id.

Return type:

PurchaseInvoiceEvidenceResult

Raises:

PurchaseInvoiceEvidenceNotFoundError – if no matching record exists.

remove(*, bucket_id, evidence_id, actor='cli')[source]

Remove an evidence record from a bucket.

Loads the bucket catalogue, finds the record, removes it from the in-memory list, persists the updated encrypted bucket-local catalogue in secure-object storage, and emits a PURCHASE_INVOICE_EVIDENCE_DETACHED audit event.

Parameters:

  • bucket_id (str) – Ledger bucket containing the record.

  • evidence_id (str) – Id of the record to remove.

  • actor (str) – Identifier stamped on the audit event.

Returns:

Carrying the removed record and the audit event id.

Return type:

PurchaseInvoiceEvidenceResult

Raises:

PurchaseInvoiceEvidenceNotFoundError – if no matching record exists.