cadrumo.application.ledger.models module

Strict Pydantic contracts for manual ledger transaction workflows.

class ManualLedgerTransactionCommand(**data)

Bases: _ManualLedgerTransactionInput

Backend command for creating or updating one manual ledger transaction.

Parameters:

data (Any)

bucket_id: BucketId

booked_date: date

value_date: date | None

amount: Decimal

currency: IsoCurrencyCode

direction: TransactionDirection

counterparty: _LedgerOptionalText

description: str

business_classification: BusinessClassification

business_pct: Decimal | None

category_id: _LedgerOptionalText

taxable_base: Decimal | None

iva_rate: Decimal | None

iva_amount: Decimal | None

recargo_amount: Decimal | None

irpf_category: _LedgerOptionalText

m210_income_classification: M210IncomeClassification | None

usage_ratio_id: _LedgerOptionalText

prorrata_reference: _LedgerOptionalText

purchase_invoice_evidence_id: _LedgerOptionalText

attachment_ids: tuple[str, ...]

notes: str

iva_category: IvaCategory | None

deduction_fact_kind: IvaDeductionFactKind | None

Operator-declared M303 deduction source. The taxonomy is documented as non-inferable, so it is carried from the operator rather than derived from the category, the direction, or the counterparty country.

counterparty_country: str | None

counterparty_identification_state: EUMemberState | None

art_104_tres_exclusion: Art104TresExclusion | None

input_classification: InputClassification | None

prorrata_sector_id: str | None

actor: str

source_command: str

idempotency_key: _LedgerOptionalText

classified_by_override: _LedgerOptionalText

source_jurisdiction: str | None

group_label: str | None

class ManualLedgerTransactionPatch(**data)

Bases: _ManualLedgerTransactionInput

Typed partial update for one bucket-scoped ledger transaction.

Parameters:

data (Any)

booked_date: date | None

value_date: date | None

amount: Decimal | None

currency: IsoCurrencyCode | None

direction: TransactionDirection | None

counterparty: _LedgerOptionalText

description: _LedgerOptionalText

business_classification: BusinessClassification | None

business_pct: Decimal | None

category_id: _LedgerOptionalText

taxable_base: Decimal | None

iva_rate: Decimal | None

iva_amount: Decimal | None

recargo_amount: Decimal | None

irpf_category: _LedgerOptionalText

m210_income_classification: M210IncomeClassification | None

usage_ratio_id: _LedgerOptionalText

prorrata_reference: _LedgerOptionalText

purchase_invoice_evidence_id: _LedgerOptionalText

attachment_ids: tuple[str, ...] | None

notes: _LedgerOptionalText

iva_category: IvaCategory | None

deduction_fact_kind: IvaDeductionFactKind | None

counterparty_country: str | None

counterparty_identification_state: EUMemberState | None

source_jurisdiction: str | None

group_label: _LedgerOptionalGroupLabel

class ManualLedgerTransactionResult(**data)

Bases: BaseModel

Backend result for a persisted manual ledger transaction mutation.

stale_finalized_revisions is populated only by an evidence-only attachment that landed on a row a finalized revision cites. Those revisions bundled their ledger evidence BEFORE the attachment, so their frozen bundles no longer show the proof the row now carries: the operator must recalculate for the evidence to reach a draft. The field is structured provenance for the caller to project into an operator notice, never a refusal.

Parameters:

data (Any)

ref: BucketTransactionRef

transaction: Transaction

bucket_event_ids: tuple[str, ...]

stale_finalized_revisions: tuple[LedgerRemovalBlocker, ...]

class LedgerTransactionPayload(**data)

Bases: _LedgerCountryCodeModel

Canonical read projection for one ledger transaction.

Parameters:

data (Any)

transaction_id: TransactionId

date: IsoDateText

booked_date: IsoDateText

value_date: str | None

amount: str

currency: IsoCurrencyCode

direction: str

counterparty: str

description: str

business_classification: str

business_pct: str | None

category_id: str | None

taxable_base: str | None

iva_rate: str | None

iva_amount: str | None

iva_category: str | None

counterparty_country: str | None

counterparty_identification_state: str | None

irpf_category: str | None

m210_income_classification: M210IncomeClassification | None

usage_ratio_id: str | None

prorrata_reference: str | None

purchase_invoice_evidence_id: str | None

invoice_id: str | None

attachment_ids: tuple[str, ...]

notes: str

lifecycle_state: str

classified_by: str

classified_at: str | None

classification_reason: str

classification_confidence: str | None

source_jurisdiction: str | None

value_in_eur: str | None

fx_rate: str | None

created_at: str

modified_at: str

class LedgerTransactionReviewPayload(**data)

Bases: LedgerTransactionPayload

Canonical read projection for one ledger transaction plus review status.

Extends LedgerTransactionPayload with the derived operator review_status so the two projections share one field set, validator, and config; review_status serialises after the inherited fields.

Parameters:

• transaction_id (TransactionId)

• date (Annotated[str, StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=None, min_length=10, max_length=10, pattern=None, ascii_only=None)])

• booked_date (Annotated[str, StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=None, min_length=10, max_length=10, pattern=None, ascii_only=None)])

• value_date (str | None)

• amount (Annotated[str, MinLen(min_length=1)])

• currency (Annotated[str, BeforeValidator(func=~cadrumo.core.parsing.codes.normalise_iso_4217_currency, json_schema_input_type=PydanticUndefined)])

• direction (Annotated[str, MinLen(min_length=1)])

• counterparty (str)

• description (Annotated[str, MinLen(min_length=1)])

• business_classification (Annotated[str, MinLen(min_length=1)])

• business_pct (str | None)

• category_id (str | None)

• taxable_base (str | None)

• iva_rate (str | None)

• iva_amount (str | None)

• iva_category (str | None)

• counterparty_country (str | None)

• counterparty_identification_state (str | None)

• irpf_category (str | None)

• m210_income_classification (M210IncomeClassification | None)

• usage_ratio_id (str | None)

• prorrata_reference (str | None)

• purchase_invoice_evidence_id (str | None)

• invoice_id (str | None)

• attachment_ids (tuple[str, ...])

• notes (str)

• lifecycle_state (Annotated[str, MinLen(min_length=1)])

• classified_by (Annotated[str, MinLen(min_length=1)])

• classified_at (str | None)

• classification_reason (str)

• classification_confidence (str | None)

• source_jurisdiction (str | None)

• value_in_eur (str | None)

• fx_rate (str | None)

• created_at (str)

• modified_at (str)

review_status: LedgerReviewStatus

class LedgerTransactionResultPayload(**data)

Bases: BaseModel

Canonical projection for a single ledger mutation/read result.

Parameters:

data (Any)

bucket_id: BucketId

transaction_id: TransactionId

review_status: LedgerReviewStatus

transaction: LedgerTransactionPayload

class LedgerTransactionTrackingPayload(**data)

Bases: BaseModel

Durable event lineage fields for one ledger transaction.

Parameters:

data (Any)

transaction_id: TransactionId

created_event_id: str | None

evidence_provenance: tuple[TransactionEvidenceProvenanceEntry, ...]

edit_lineage: tuple[TransactionEditLineageEntry, ...]

lifecycle_state: str

lifecycle_lineage: tuple[TransactionLifecycleLineageEntry, ...]

class SplitChildCommand(**data)

Bases: BaseModel

One slice of an N-way split.

Variables:

• amount – Non-negative magnitude Decimal in the parent’s currency; direction is inherited from the parent (the split builder copies parent.direction onto every child).

• description – Non-blank narrative for this slice.

• counterparty – Optional override; falls back to the parent’s counterparty when None.

• booked_date – Optional override; falls back to the parent’s booked_date.

• value_date – Optional override; falls back to the parent’s value_date.

Parameters:

data (Any)

amount: Decimal

description: str

counterparty: str | None

booked_date: date | None

value_date: date | None

class SplitTransactionResult(**data)

Bases: BaseModel

Backend result of a successful N-way split.

Parameters:

data (Any)

bucket_id: BucketId

parent_transaction_id: TransactionId

split_group_id: Hex64Str

child_transaction_ids: tuple[str, ...]

parent_transaction: Transaction

child_transactions: tuple[Transaction, ...]

bucket_event_id: Hex64Str

class MergeTransactionsResult(**data)

Bases: BaseModel

Backend result of a successful split re-merge.

Parameters:

data (Any)

bucket_id: BucketId

split_group_id: Hex64Str

parent_transaction_id: TransactionId

merged_transaction_id: TransactionId

source_child_ids: tuple[str, ...]

merged_transaction: Transaction

parent_transaction: Transaction

bucket_event_id: Hex64Str

class LedgerImportOperationResult(**data)

Bases: BaseModel

Backend result for an imported ledger transaction batch.

Parameters:

data (Any)

summary: ImportSummary

import_batch_id: Hex64Str | None

bucket_event_ids: tuple[str, ...]

class LedgerSourceValidationReport(**data)

Bases: BaseModel

Backend read model for source-provider validation details.

Parameters:

data (Any)

valid: bool

warnings: tuple[str, ...]

encoding: str | None

dialect: str | None

class LedgerSourceVerificationReport(**data)

Bases: BaseModel

Backend read model for optional source-file verification.

Parameters:

data (Any)

requested: bool

path: str | None

sha256: str | None

class LedgerImportDiagnosticReport(**data)

Bases: BaseModel

Backend read model for one persisted or dry-run import diagnostic.

Parameters:

data (Any)

kind: DiagnosticKind

severity: DiagnosticSeverity

message: DiagnosticMessage

source_path: str | None

source_locator: str | None

affected_transaction_ids: tuple[str, ...]

class LedgerSourceImportCommand(**data)

Bases: BaseModel

Backend command for importing ledger rows from an operator source file.

Parameters:

data (Any)

bucket_id: BucketId | None

path: Path

provider: str

dry_run: bool

verify: bool

source: Path | None

period: Period | None

actor: str

source_command: str

class LedgerSourceImportResult(**data)

Bases: BaseModel

Backend result for source-file import diagnostics and persistence.

Parameters:

data (Any)

rows: NonNegativeInt

imported: NonNegativeInt

skipped: NonNegativeInt

likely_duplicates: int

dry_run: bool

verify: bool

period: Period | None

bucket_id: BucketId | None

import_batch_id: str | None

bucket_event_ids: tuple[str, ...]

imported_transaction_refs: tuple[BucketTransactionRef, ...]

skipped_transaction_refs: tuple[BucketTransactionRef, ...]

likely_duplicate_transaction_refs: tuple[BucketTransactionRef, ...]

validations: tuple[LedgerSourceValidationReport, ...]

sources: tuple[LedgerSourceVerificationReport, ...]

diagnostics: tuple[LedgerImportDiagnosticReport, ...]

class LedgerReviewQuery(**data)

Bases: BaseModel

Backend query for operator review rows in one bucket ledger.

Parameters:

data (Any)

bucket_id: BucketId

period: Period | None

status: str | None

issue: str | None

import_id: str | None

classification: str | None

text: str | None

direction: str | None

transaction_id: TransactionId | None

class LedgerReviewRow(**data)

Bases: BaseModel

Backend projection for one ledger review row.

id is the content-addressed ledger transaction identity, not merely a 64-character string: it is the value the operator passes back to address the row, so an id that is the right length but not a digest names nothing.

Parameters:

data (Any)

id: TransactionId

date: IsoDateText

amount: str

description: str

status: str

transaction: LedgerTransactionPayload | None

class LedgerReviewQueryResult(**data)

Bases: BaseModel

Backend result for operator ledger review queries.

Parameters:

data (Any)

bucket_id: BucketId

rows: tuple[LedgerReviewRow, ...]

filters: tuple[str, ...]

class LedgerStatusReport(**data)

Bases: BaseModel

Backend read model summarizing one bucket’s ledger transaction state.

Parameters:

data (Any)

bucket_id: BucketId

business_income_total: str

business_expense_total: str

business_net_total: str

total_count: NonNegativeInt

active_count: NonNegativeInt

archived_count: NonNegativeInt

stashed_count: NonNegativeInt

split_count: int

pending_review_count: NonNegativeInt

reviewed_count: NonNegativeInt

skipped_count: NonNegativeInt

period: Period | None

checked_transaction_count: int

readiness_issue_count: int

unconverted_currency_count: int

ready: bool | None

class LedgerRemovalBlocker(**data)

Bases: BaseModel

Finalized modelo revision that prevents ledger transaction removal.

Parameters:

data (Any)

work_unit_id: WorkUnitId

calculation_revision_id: CalculationRevisionId

revision_state: str

modelo: str

filing_year: FilingYear

period: str

class LedgerTransactionRemovalReport(**data)

Bases: BaseModel

Backend report for one bucket-local ledger transaction removal.

Parameters:

data (Any)

bucket_id: BucketId

transaction_id: TransactionId

removed: bool

dry_run: bool

actor: str

reason: str

cascaded_purchase_invoice_evidence_ids: tuple[str, ...]

cascaded_attachment_ids: tuple[str, ...]

blocking_modelo_references: tuple[LedgerRemovalBlocker, ...]

stale_draft_revision_references: tuple[LedgerRemovalBlocker, ...]

bucket_event_ids: tuple[str, ...]

class LedgerCatalogueResetReport(**data)

Bases: BaseModel

Backend report for a protected bucket-local ledger catalogue reset.

Parameters:

data (Any)

bucket_id: BucketId

removed_transaction_ids: tuple[str, ...]

reset: bool

dry_run: bool

actor: str

reason: str

cascaded_purchase_invoice_evidence_ids: tuple[str, ...]

cascaded_attachment_ids: tuple[str, ...]

blocking_modelo_references: tuple[LedgerRemovalBlocker, ...]

stale_draft_revision_references: tuple[LedgerRemovalBlocker, ...]

bucket_event_ids: tuple[str, ...]

class LedgerExportCommand(**data)

Bases: BaseModel

Backend command for exporting one bucket’s canonical ledger rows.

Parameters:

data (Any)

bucket_id: BucketId

export_format: ExportSerializationFormat

include_inactive: bool

output_path: Path | None

period: Period | None

actor: str

source_command: str

class BulkClassifyRow(**data)

Bases: BaseModel

One row from a ledger classify --file CSV input file.

Required columns: transaction_id, classification. Optional columns: category_id, business_pct, usage_ratio_id, taxable_base, iva_rate, iva_amount, iva_category, irpf_category. Unknown column names are rejected pre-persistence to protect against silent field mis-mapping. The IVA facts (taxable_base, iva_rate, iva_amount) are typed Decimal exactly as the single-classify path coerces them, so a malformed value reds the row rather than coercing silently.

Parameters:

data (Any)

transaction_id: TransactionIdReference

What the operator typed, not a resolved id. A CSV names rows the way a terminal does – by the short display id ledger list prints – and typing this as the resolved 64-character id refused every such row at parse time, before resolve_transaction_id in the apply path could resolve it. The single-row verb has always accepted a prefix.

classification: BusinessClassification

category_id: str | None

business_pct: Decimal | None

usage_ratio_id: str | None

taxable_base: Decimal | None

iva_rate: Decimal | None

iva_amount: Decimal | None

iva_category: IvaCategory | None

irpf_category: str | None

class BulkClassifyFailure(**data)

Bases: BaseModel

One failed row from a ledger classify --file operation.

Parameters:

data (Any)

row_index: NonNegativeInt

transaction_id: str

reason: str

class BulkClassifyResult(**data)

Bases: BaseModel

Aggregate result for a ledger classify --file operation.

Uses partial-success semantics matching the ledger import pattern: all parseable rows that pass validation are applied; failures are collected into failures and reported without aborting the batch.

Parameters:

data (Any)

total: NonNegativeInt

applied: NonNegativeInt

skipped: NonNegativeInt

failures: tuple[BulkClassifyFailure, ...]

bucket_event_ids: tuple[str, ...]

class ApplyRulesAppliedRow(**data)

Bases: BaseModel

One successfully rule-applied transaction from ledger rule apply.

Parameters:

data (Any)

transaction_id: TransactionId

matched_rule_id: str

classification: BusinessClassification

class ApplyRulesResult(**data)

Bases: BaseModel

Aggregate result for a ledger rule apply operation.

Parameters:

data (Any)

rules_evaluated: NonNegativeInt

transactions_scanned: NonNegativeInt

matched: NonNegativeInt

skipped_already_classified: NonNegativeInt

no_match: NonNegativeInt

applied: tuple[ApplyRulesAppliedRow, ...]

bucket_event_ids: tuple[str, ...]

class LedgerExportRow(**data)

Bases: BaseModel

One serialized ledger transaction row exported from the canonical catalogue.

Parameters:

data (Any)

bucket_id: BucketId

transaction_id: TransactionId

lifecycle_state: NonEmptyStr

booked_date: IsoDateText

value_date: str

effective_date: IsoDateText

amount: NonEmptyStr

currency: IsoCurrencyCode

direction: NonEmptyStr

counterparty: str

description: NonEmptyStr

business_classification: NonEmptyStr

business_pct: str

category_id: str

taxable_base: str

iva_rate: str

iva_amount: str

iva_category: str

counterparty_country: str

counterparty_identification_state: str

irpf_category: str

usage_ratio_id: str

prorrata_reference: str

purchase_invoice_evidence_id: str

attachment_ids: str

notes: str

created_by: str

created_source_command: str

source_jurisdiction: str

value_in_eur: str

fx_rate: str

class LedgerExportResult(**data)

Bases: BaseModel

Backend result for an exported canonical ledger transaction snapshot.

Parameters:

data (Any)

bucket_id: BucketId

export_id: Hex64Str

export_format: ExportSerializationFormat

media_type: str

filename_extension: str

row_count: NonNegativeInt

byte_size: NonNegativeInt

sha256: ContentDigest

fieldnames: tuple[str, ...]

rows: tuple[LedgerExportRow, ...]

payload: bytes

bucket_event_ids: tuple[str, ...]