aeat.application.invoices._importing module

Application service for importing invoice records.

import_invoices_from_path() loads the persisted InvoiceCatalogue from the InvoiceCatalogueRepository, merges the parsed invoices, and writes the updated catalogue back.

class InvoiceRowPayload

Bases: TypedDict

Typed shape for a single decoded invoice row from JSON input.

All fields are optional at the decode boundary because JSON sources may omit fields that are defaulted downstream in parse_invoice_payload(). The downstream coercion stage fills in application defaults before Invoice validation runs.

invoice_id: NotRequired[str]

bucket_id: NotRequired[str]

kind: NotRequired[str]

currency: NotRequired[str]

invoice_number: NotRequired[str]

issued_at: NotRequired[str]

counterparty_name: NotRequired[str]

counterparty_tax_id: NotRequired[str]

counterparty_country: NotRequired[str]

payment_status: NotRequired[str]

lines: NotRequired[list[Any]]

base_total: NotRequired[str]

iva_total: NotRequired[str]

grand_total: NotRequired[str]

linked_transaction_ids: NotRequired[list[str]]

notes: NotRequired[str]

iva_category: NotRequired[str]

operation_type: NotRequired[str]

oss_ioss_regime: NotRequired[str]

oss_transaction_kind: NotRequired[str]

retention_rate: NotRequired[str]

retention_amount: NotRequired[str]

payment_id: NotRequired[str]

class InvoiceImportResult(**data)

Bases: BaseModel

Result DTO returned by invoice import application services.

Parameters:

• rows (int)

• imported (int)

• skipped (int)

• dry_run (bool)

• catalogue (InvoiceCatalogue | None)

rows: int

imported: int

skipped: int

dry_run: bool

catalogue: InvoiceCatalogue | None

parse_invoice_payload(raw, *, default_kind)

Parse JSON invoice payloads into validated Invoice models.

Return type:

tuple[Invoice, ...]

Parameters:

• raw (str)

• default_kind (InvoiceKind | str)

merge_invoice_import(catalogue, invoices)

Merge imported invoices into catalogue without duplicating IDs.

Parameters:

• catalogue (InvoiceCatalogue) – The InvoiceCatalogue to merge the imported invoices into.

• invoices (Sequence[Invoice]) – Sequence of Invoice rows produced by an importer; rows whose invoice_id already exists in catalogue are skipped to preserve catalogue identity.

Return type:

InvoiceImportResult

Returns an InvoiceImportResult with the updated catalogue and counts of imported and skipped invoices.

import_invoices_from_path(path, *, kind, dry_run=False, repository=None)

Import invoices from path through the secure invoice repository.

Returns an InvoiceImportResult.

Return type:

InvoiceImportResult

Parameters:

• path (Path)

• kind (InvoiceKind | str)

• dry_run (bool)

• repository (InvoiceCatalogueRepositoryProtocol | None)