cadrumo.application.transactions.import_diagnostics 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 routes through classify_import_row(), the same verdict the persisting ledger import path consumes, so the diagnostics and the persistence path cannot disagree about what a row is. They did: this module once counted a fingerprint repeated within one file as skipped while the persisting path imported both rows, so --verify reported a duplicate the import would never skip. An intra-batch repeat is therefore an advisory here, not a skip.

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:

data (Any)

imported_count: int
skipped_count: int
diagnostics: tuple[LedgerImportDiagnostic, ...]
import_ledger_with_diagnostics(source_path, raw_transactions, existing_catalogue, import_fingerprints, original_source_path=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.

  • import_fingerprints (Iterable[str]) – One import fingerprint per raw row, derived by the caller with the flow direction the provider read at the parse boundary. Required rather than defaulted: a fingerprint derived here would carry no direction, and derive_import_fingerprint() substitutes the literal UNSPECIFIED discriminator for a missing one, which can never equal a stored direction-qualified fingerprint. Dedup would then fail open and read every row as new.

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

Return type:

LedgerImportResult

Returns:

An immutable LedgerImportResult with finding diagnostics.