aeat.application.transactions._import module

Orchestration layer for ledger imports with diagnostics.

import_ledger_with_diagnostics() accepts an existing TransactionCatalogue and an iterable of RawTransaction rows. It emits structured diagnostics for parser-empty, duplicate, calendar-gap, and original-file checks during import verification.

Duplicate detection uses the stable derive_import_fingerprint() identity so the dry-run diagnostics match the persistence path used by ledger import actions.

class LedgerImportResult(**data)[source]

Bases: BaseModel

Return value of an orchestrated ledger import with diagnostics.

imported_count and skipped_count reflect the previewed outcome for the supplied RawTransaction rows; diagnostics carries the structured LedgerImportDiagnostic records explaining parser, duplicate, gap, or original-file findings.

Parameters:
imported_count: int
skipped_count: int
diagnostics: tuple[LedgerImportDiagnostic, ...]
import_ledger_with_diagnostics(source_path, raw_transactions, existing_catalogue, original_source_path=None, import_fingerprints=None)[source]

Evaluate an imported stream against the four diagnostic checks.

Parameters:

  • source_path (Path) – Path of the file being imported.

  • raw_transactions (Iterable[RawTransaction]) – Unmerged raw transactions emitted by the provider.

  • existing_catalogue (TransactionCatalogue) – The current TransactionCatalogue used for import-fingerprint duplicate detection.

  • original_source_path (Path | None) – Optional original file path to record when it is present on disk.

  • import_fingerprints (Iterable[str] | None) – Optional per-row import fingerprints derived by the caller from parse-boundary facts such as transaction direction.

Return type:

LedgerImportResult

Returns:

An immutable LedgerImportResult with finding diagnostics.