cadrumo.application.ledger.actions_common module

Shared ledger action helpers for repositories, events, and guards.

This module validates injected ledger persistence ports; builds BucketEvent audit entries; mutates TransactionCatalogue and InvoiceCatalogue snapshots atomically; and verifies evidence, attachment, usage-ratio, and finalized-modelo blockers for the public ledger action services.

require_repository(repository, *, reason)[source]

Require a port selected by an outer composition root.

Ledger application services deliberately do not choose persistence implementations. Failing before any read/write preserves the prior all-or-nothing action semantics while making an unwired production caller explicit.

Return type:

TypeVar(RepositoryT)

Parameters:

  • repository (RepositoryT | None)

  • reason (str)

resolve_transaction_repository(*, bucket_id, repository)[source]
Return type:

TransactionCatalogueCoCommitWriterProtocol

Parameters:
resolve_invoice_repository(*, bucket_id, repository)[source]
Return type:

InvoiceCatalogueCoCommitWriterProtocol

Parameters:
resolve_bucket_event_repository(*, bucket_id, repository)[source]
Return type:

BucketEventHistoryCoCommitWriterProtocol

Parameters:
require_actor(value, *, operation)[source]
Return type:

str

Parameters:
require_source_command(value, *, operation)[source]
Return type:

str

Parameters:
normalise_attachment_patch_ids(attachment_ids)[source]
Return type:

tuple[str, ...]

Parameters:

attachment_ids (tuple[str, ...])

merge_identifier_tuple(existing, incoming)[source]
Return type:

tuple[str, ...]

Parameters:
required_patched(patch, patch_fields, field, fallback)[source]

Return patch.<field> when in patch_fields and non-null; otherwise the fallback.

Raises TransactionValidationError when the field is set on the patch but explicitly nulled - the patch contract forbids resetting a required value to None. Generic in T so the caller’s field type flows through to the assignment site (type checkers see the concrete type at the use point, not object).

Return type:

TypeVar(T)

Parameters:
optional_patched(patch, patch_fields, field, fallback)[source]

Return patch.<field> when in patch_fields (None allowed); otherwise the fallback.

Generic in T for the same reason as required_patched().

Return type:

TypeVar(T)

Parameters:
blocking_modelo_references(*, bucket_id, transaction_ids, work_unit_repository, calculation_repository)[source]

Return finalized modelo revisions that still depend on these transactions.

Return type:

tuple[LedgerRemovalBlocker, ...]

Parameters:
draft_revision_advisories(*, bucket_id, transaction_ids, work_unit_repository, calculation_repository)[source]

Collect DRAFT (BORRADOR) revisions still citing the wanted transaction ids.

Removal proceeds for draft-cited rows, but the draft’s source_transaction_ids will still assert the removed row’s income/expense on the next verify/file. These advisory rows name each affected draft so the operator can recalculate it (no-silent-under-declaration). Correctness rides on the live revision-catalogue scan, never the derived participation index (aeat-ledger-contract).

Return type:

tuple[LedgerRemovalBlocker, ...]

Parameters:
blockers_by_source_transaction_id(*, bucket_id, work_unit_repository, calculation_repository)[source]

Map each source transaction id to the finalized-modelo blockers referencing it.

Computed once so a batch edit can look up the finalized-modelo guard per row without reloading the work-unit and calculation repositories on every row (the load-once half of the bulk_classify_from_csv batching contract).

Return type:

dict[str, tuple[LedgerRemovalBlocker, ...]]

Parameters:
transaction_modelo_source_ids(transaction)[source]
Return type:

tuple[str, ...]

Parameters:

transaction (Transaction)

catalogue_modelo_source_ids(catalogue)[source]
Return type:

tuple[str, ...]

Parameters:

catalogue (TransactionCatalogue)

is_evidence_only_command(command, current)[source]

Return whether command adds evidence provenance and changes nothing else.

An evidence-only mutation is exempt from the finalized-modelo write guard, because it cannot disturb any finalized revision that cites the row. Three independent project contracts establish that:

  • derive_transaction_id() hashes the provider identity, effective value date, amount, and narrative only, so an evidence-only edit re-derives the SAME id. A finalized revision’s source_transaction_ids citation therefore keeps resolving.

  • The ledger filing snapshot’s fingerprint field set (the canonical “tax facts” of a row) excludes both evidence fields, so row_fingerprint and the revision’s snapshot_fingerprint are unchanged and the revision’s LedgerFilingSnapshot still reconciles against the live catalogue.

  • A finalized revision’s casilla values and bundled ledger evidence are frozen persisted snapshots; this write touches the transaction catalogue alone.

The exemption is deliberately the narrowest that unblocks the operator: it requires at least one evidence field to differ AND every other persisted field to match, so a value-affecting edit smuggled alongside an attachment still meets the guard. Adding evidence DOES change what a later recalculation would compute (an incoming row with purchase evidence becomes a Renta refund, and a resolved invoice’s base/IVA override the row’s own), which is why the caller reports the cited revisions as stale rather than silently accepting the write.

Return type:

bool

Parameters:
raise_finalized_modelo_blocked(*, operation, transaction_ids, blockers)[source]
Return type:

None

Parameters:
transaction_catalogue_object_id(bucket_id)[source]
Return type:

str

Parameters:

bucket_id (str)

verify_evidence_references(command, *, transaction_id, invoice_repository, attachment_store, evidence_records=())[source]
Return type:

None

Parameters:
resolve_attachment_store(attachment_store)[source]

Require the attachment store selected by the outer composition root.

Return type:

AttachmentStoreProtocol

Parameters:

attachment_store (AttachmentStoreProtocol | None)

verify_usage_ratio_reference(command, *, usage_ratio_profile)[source]
Return type:

None

Parameters:
optional_decimal(value)[source]
Return type:

str

Parameters:

value (Decimal | None)

display_decimal(value)[source]
Return type:

str

Parameters:

value (Decimal)

normalise_timestamp(value)[source]
Return type:

datetime

Parameters:

value (datetime | None)

upsert_transaction(catalogue, transaction)[source]
Return type:

TransactionCatalogue

Parameters:
replace_transaction(catalogue, *, old_transaction_id, replacement)[source]
Return type:

TransactionCatalogue

Parameters:
remove_transaction(catalogue, *, transaction_id)[source]
Return type:

TransactionCatalogue

Parameters:
require_transaction(catalogue, transaction_id)[source]
Return type:

Transaction

Parameters:
mutation_signature(transaction)[source]

Fold every operator-settable ledger field into one comparable tuple.

This is what decides whether an edit changed anything: two equal signatures make the update a no-op, and the caller refuses it with “manual ledger update must change at least one ledger field”. A persisted field missing from here is therefore not merely uncompared — an edit touching ONLY that field is rejected and the operator’s correction is dropped.

It must stay key-for-key with _transaction_idempotency_fields(), minus classified_by alone. That exclusion is deliberate: classified_by is provenance a write stamps, not a value the operator set, so comparing it would make every re-classification look like a change and defeat the guard. Re-applying an identical classification on purpose is what --reaffirm is for.

recargo_amount and deduction_fact_kind were absent, and both are Modelo 303 figures rather than provenance: an operator correcting a recargo de equivalencia, or the deduction source behind an input-IVA claim, was told nothing had changed.

Return type:

tuple[object, ...]

Parameters:

transaction (Transaction)

command_matches_current(command, current)[source]

Return True when a command would produce no observable change against the stored transaction.

Used to detect re-affirmation patches (operator supplies the same business_classification the record already carries) so the caller can treat them as confirmed no-ops instead of raising a mutation-required error.

Return type:

bool

Parameters:
build_ledger_bucket_event(*, bucket_id, event_type, occurred_at, actor, object_id, payload, object_type=BucketEventObjectType.LEDGER_TRANSACTION)[source]

Derive one ledger BucketEvent without persisting it.

Delegates to the domain builder rather than restating the envelope, so the ledger cannot drift from the canonical event shape. Supplies the ledger’s _BUCKET_EVENT_PAYLOAD_VERSION and its default object type.

Return type:

BucketEvent

Parameters:
save_transaction_catalogue_and_events(*, transaction_repository, event_repository, catalogue, events)[source]
Return type:

None

Parameters:
save_transaction_catalogue_invoices_and_events(*, transaction_repository, invoice_repository, event_repository, transaction_catalogue, invoice_catalogue, events)[source]
Return type:

None

Parameters:
primary_lineage_event_id(events)[source]
Return type:

str

Parameters:

events (tuple[BucketEvent, ...])

derive_evidence_event_ids(events)[source]
Return type:

dict[tuple[str, str], str]

Parameters:

events (tuple[BucketEvent, ...])

build_manual_ledger_result(bucket_id, transaction, bucket_event_ids, *, stale_finalized_revisions=())[source]
Return type:

ManualLedgerTransactionResult

Parameters: