cadrumo.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.

A PurchaseInvoiceEvidence record is the MIDDLE tier of the three-rung evidence progression, and owns no bytes of its own:

  1. Attachment owns byte custody. It is strictly content-addressed (attachment_id == sha256 of the stored bytes), immutable, and carries no fiscal figures. aeat app ledger attach and aeat app ledger doclink link one directly to a transaction.

  2. PurchaseInvoiceEvidence is an operator-registered CLAIM ABOUT one such byte payload: a mutable record whose supplier, invoice number, invoice date, and IVA figures are all OPTIONAL, because a scan whose text layer yields nothing is still valid evidence. Its evidence_id is a metadata digest, not a content digest, so several records may describe one byte payload; the bytes themselves are stored once, as the Attachment written at add time and read back through attachment_id.

  3. Invoice is the CONFIRMED fiscal document, whose counterparty name, tax id, country, totals, currency, and lines are all REQUIRED. aeat app ledger evidence confirm promotes tier 2 to tier 3 once the operator supplies or accepts those figures.

The tiers are a permissiveness ladder, not three ways of saying one thing: each rung requires strictly more than the one below it, so none can absorb another without either dropping fiscal fields or refusing evidence the tier below legitimately accepts.

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 and the source bytes are written through the required LedgerEvidencePorts custody capability. 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
class PurchaseInvoiceEvidence(**data)[source]

Bases: BaseModel

One persisted purchase invoice evidence record.

Parameters:

data (Any)

evidence_id: str
bucket_id: BucketId
source_path: str
source_sha256: ContentDigest
attachment_id: Hex64Str
media_kind: MediaKind
supplier: str | None
invoice_number: str | None
invoice_date: str | None
taxable_base: NonNegativeDecimal | None
iva_rate: Percentage | None
iva_amount: NonNegativeDecimal | 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 cadrumo.domain.transactions.models.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:
derive_keyed_purchase_invoice_evidence_id(*, bucket_id, idempotency_key)[source]

Return a CLOCK-FREE evidence id for a caller-supplied idempotency key.

The keyless derivation above deliberately folds created_at plus a disambiguator, and that is not an oversight to correct: two evidence records for the same file are a legitimate case – the same invoice PDF can be attached twice as two distinct pieces of evidence – and simply dropping the clock would silently collapse them into one. The codified rule anticipates exactly this and supplies the shape: a deliberately-additive verb documents itself as such, while a caller-supplied key provides the guarded path for callers that need one.

So this is not “drop the clock”, it is “add the key”. The id derives from the bucket and the key alone, which is what makes a retry at a different instant resolve to the same record.

Return type:

str

Parameters:

  • bucket_id (str)

  • idempotency_key (str)

class PurchaseInvoiceEvidenceDocument(**data)[source]

Bases: BaseModel

Encrypted bucket-local purchase invoice evidence catalogue.

Parameters:

data (Any)

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

data (Any)

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

Bases: BaseModel

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

Parameters:

data (Any)

record: PurchaseInvoiceEvidence
bucket_event_ids: tuple[str, ...]
class PurchaseInvoiceEvidenceService(*, ports)[source]

Bases: object

Application service for the aeat app ledger evidence verb group.

Parameters:

ports (LedgerEvidencePorts)

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', idempotency_key=None)[source]

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

Resolves source_path for byte access only, verifies the file exists, infers the MediaKind from the extension, copies the file’s bytes through the required attachment-ingestion capability (active bucket) and records the resulting content-addressed attachment_id on the record (the bytes thereafter live only in secure storage). The persisted source_path is the argv-faithful path the operator supplied, never the machine-absolutized form, so it is a stable provenance breadcrumb rather than a machine-dependent one. 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 whose content-addressed id derives from the source digest and stable metadata, never the path.

Parameters:

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

  • source_path (str | Path) – Local path to a PDF or image file. A str (the raw operator argv) is echoed onto the record verbatim — separators are preserved exactly, never OS-normalized — so the persisted path and envelope are identical across platforms (a forward-slash relative path stays forward-slash on Windows). A Path is accepted for programmatic callers and stringified for the echo. Byte access always resolves the path regardless.

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

  • idempotency_key (str | None) – Caller-supplied retry key. When supplied the record id is derived CLOCK-FREE from it, so a retry at a different instant resolves to the same record: a matching re-add returns the existing record as a guarded no-op with no second bucket event and no re-stamped timestamp, and a same-key re-add whose content differs refuses naming the divergent fields. When omitted the verb stays deliberately ADDITIVE – two attachments of one file are two distinct pieces of evidence, and collapsing them would be its own defect.

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.

emit_evidence_event(*, event_repository, bucket_id, event_type, evidence_id, actor, occurred_at, payload)

Record one evidence transition through the shared emission primitive.

This used to derive the id, build the event, append it and save the catalogue itself – the exact sequence emit_bucket_event() documents as the one every emitting domain must share. Re-deriving it cost more than duplication: the shared primitive appends through the catalogue’s revision guard, and a bare load-append-save does not, so two evidence attachments landing together discarded one another’s audit entry. Content-addressed events make that loss invisible after the fact – every survivor is intact and the missing one leaves no gap – so the trail still read as complete.

The only things this surface fixes are the object kind and the payload version; everything else is the caller’s.

Return type:

str

Parameters: