aeat.domain.calculations.registry._invoice_bindings module

Invoice-shaped registry binding helpers.

class InvoiceObservation(**data)[source]

Bases: BaseModel

One factual line from the user’s invoice ledger.

The fields are scoped to the facts every IVA modelo needs to classify a transaction. base_amount carries the taxable base; invoice_total_amount carries the gross invoice total for modelos such as M347 whose declaration floor is not the taxable-base amount. intracommunity_clave follows the AEAT clave-de-operacion enum (E, M, H, A, T, S, I, R, D, C). iva_regime is open-ended so domestic-IVA modelos can carry their regime classification alongside.

Parameters:

  • invoice_id (str)

  • source_kind (BindingSourceKind)

  • party_tax_id (str)

  • country_code (str)

  • transaction_date (date)

  • base_amount (Decimal)

  • invoice_total_amount (Decimal | None)

  • iva_regime (str | None)

  • intracommunity_clave (str | None)

  • is_rectification (bool)

  • rectified_year (int | None)

  • rectified_period (str | None)

  • rectified_base_previous (Decimal | None)

  • party_legal_name (str | None)

invoice_id: str
source_kind: BindingSourceKind
party_tax_id: str
country_code: str
transaction_date: date
base_amount: Decimal
invoice_total_amount: Decimal | None
iva_regime: str | None
intracommunity_clave: str | None
is_rectification: bool
rectified_year: int | None
rectified_period: str | None
rectified_base_previous: Decimal | None
class InvoiceObservationRequirement(**data)[source]

Bases: BaseModel

Invoice-fact slice declared by one or more invoice-source bindings.

Modelo runtimes use this introspection to ask the invoice ledger for the minimal set of observations the bindings need.

Parameters:

  • binding_ids (tuple[BindingId, ...])

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

  • rectification_scope (Literal['only_rectifications', 'exclude_rectifications', 'any'])

  • iva_regime (str | None)

binding_ids: tuple[BindingId, ...]
claves: tuple[str, ...]
rectification_scope: _RectificationScope
iva_regime: str | None
invoice_binding_requirements(revision)[source]

Return invoice ledger slices needed by revision’s invoice bindings.

Parameters:

revision (ModeloRevision) – The ModeloRevision whose invoice bindings are inspected.

Return type:

tuple[InvoiceObservationRequirement, ...]

Returns:

Tuple of InvoiceObservationRequirement records describing each distinct invoice-fact slice the revision requires.

validate_invoice_binding_definition(binding)[source]

Validate an invoice-source binding before it reaches runtime.

Return type:

None

Parameters:

binding (DataBindingDefinition)

validate_invoice_binding(binding)[source]

Validate an invoice-source binding at registry-build time.

Accumulating list[str] validator: validates the selector against _InvoiceSelector and lifts the invoice fact/op invariants to build time, preserving the underlying pydantic field error. This is the list[str] companion to the raise-style validate_invoice_binding_definition() (kept as a defence-in-depth resolve-time re-check).

Return type:

list[str]

Parameters:

binding (DataBindingDefinition)

validate_invoice_family_fact_and_aggregation(binding, selector, *, family_label, strict_scalar_shape)[source]

Shared invoice/counterpart fact + aggregation-op cross-invariant.

The invoice and counterpart families share one selector shape (_InvoiceSelector) and one fact set (_INVOICE_FACTS); their op/fact cross-checks were near-verbatim copies differing only in the family-name in the unsupported-fact message and in whether the invoice-only scalar-shape guards (non-row fact must not declare row_field/grouping and op 'rows' requires fact 'row_field') run. family_label selects the error wording; strict_scalar_shape toggles the invoice-only guards (the counterpart variant historically omitted them, so the flag preserves that behaviour exactly).

Return type:

None

Parameters:
resolve_invoice_family_scalar_values(revision, *, source_kinds, validate_selector, observations_for_binding)[source]

Resolve scalar bindings on a ModeloRevision for one invoice family into Decimal aggregates.

Shared core for both the invoice and counterpart scalar resolvers; the two differed only in (a) the family membership set, (b) the per-family selector validator, and (c) whether observations are filtered directly (invoice) or matched by source_kind and converted from counterpart observations. Row-producer bindings (fact == "row_field") are skipped here.

Return type:

dict[TypeAliasType, Decimal]

Parameters:
resolve_invoice_family_row_values(revision, *, source_kinds, validate_selector, observations_for_binding, cohort_by_source)[source]

Resolve row-producer bindings on a ModeloRevision for one invoice family into per-row values.

Shared core for both the invoice and counterpart row resolvers. Bindings sharing the same cohort key share one-based row indexes so that an export record with repeat = "binding_rows" correlates field values across bindings on the same row. The counterpart family adds binding.source to the cohort key (cohort_by_source = True) so a different counterpart source kind does not share rows; the invoice family does not.

Return type:

dict[tuple[TypeAliasType, int], Decimal | str]

Parameters:
resolve_invoice_binding_values(revision, observations)[source]

Resolve scalar invoice-source bindings into Decimal aggregates.

Row-producer bindings (aggregation.op == "rows") are skipped here; they are resolved by resolve_invoice_binding_row_values().

Parameters:
Return type:

dict[TypeAliasType, Decimal]

resolve_invoice_binding_row_values(revision, observations)[source]

Resolve row-producer invoice bindings into per-row indexed values.

Bindings with aggregation.op == "rows" aggregate observations into rows deterministically grouped by selector.grouping. Bindings sharing the same grouping/scope/clave-filter share row indexes, so that an export record with repeat = "binding_rows" can correlate field values across bindings on the same row. Returns a flat mapping keyed by (binding_id, row_index). Row indexes are one-based to match ModeloBindingValue.row_index.

Parameters:
Return type:

dict[tuple[TypeAliasType, int], Decimal | str]

class Modelo349OperadorClaveTotal(**data)[source]

Bases: BaseModel

One clave’s operator count and base-imponible sum, reconstructed from the operador row set.

Groups the per-operador iva-349-operador-row-* detail (the AEAT Diseño de Registros Tipo-2 “registro de operador” rows) by clave de operación and carries that clave’s distinct-operator count and summed base imponible. Mirrors WithholdingClaveBreakdown for the Modelo 349 intracommunity-operator axis.

Parameters:
clave: str
operator_count: int
base_total: Decimal
class Modelo349OperadorTotalsParity(**data)[source]

Bases: BaseModel

Totals-parity verdict between the per-operador row set and the Modelo 349 declarant summary.

Modelo 349’s declarant-summary scalar casillas (decl.numero-operadores, decl.importe-operaciones) and the per-operador-clave row-producer bindings (iva-349-operador-row-*) are resolved by two structurally INDEPENDENT code paths over the same InvoiceObservation set: resolve_invoice_binding_values() folds the observations directly into a scalar (_aggregate_invoice_binding()’s operator_count / base_sum facts), while resolve_invoice_binding_row_values() groups them into per-(country, party_tax_id, clave) rows (_build_operator_clave_rows()). Nothing in the registry cross-checks that the two paths agree, so a defect in either aggregator — or a manual-entry Modelo349OperadorRow set that omits an operator the summary already counted — would silently under- or over-declare one side without detection.

This model is the pure comparison result of that cross-check: the sum of every reconstructed operador row’s base imponible against the resolved decl.importe-operaciones value, and the count of distinct (country_code, party_tax_id, clave) operador rows against the resolved decl.numero-operadores value. is_consistent is True only when the operator-count delta is exactly zero and the base-imponible delta is within tolerance — a divergence on either axis surfaces as a loud, actionable finding (no-silent-under-declaration), never a silent pass.

Parameters:
by_clave: tuple[Modelo349OperadorClaveTotal, ...]
operator_row_total: int
operator_summary_total: int
operator_delta: int
base_row_total: Decimal
base_summary_total: Decimal
base_delta: Decimal
tolerance: Decimal
is_consistent: bool
compute_modelo_349_operador_totals_parity(revision, observations, *, operator_summary_total, base_summary_total, tolerance=Decimal('0.01'))[source]

Cross-check the per-operador row set against the resolved Modelo 349 declarant summary.

Parameters:

  • revision (ModeloRevision) – The ModeloRevision whose iva-349-operador-row-* row-producer bindings are resolved into the operador row set.

  • observations (Iterable[InvoiceObservation]) – The InvoiceObservation rows the revision’s invoice-source bindings aggregate (exclude-rectifications scope only feeds the operador row set; rectification observations are a distinct AEAT record type and are excluded from this axis by the registry’s own binding selectors).

  • operator_summary_total (Decimal) – The resolved value of casilla decl.numero-operadores, typically read from revision.casilla_values["decl.numero-operadores"].

  • base_summary_total (Decimal) – The resolved value of casilla decl.importe-operaciones, typically read from revision.casilla_values["decl.importe-operaciones"].

  • tolerance (Decimal) – Maximum absolute EUR delta on the base-imponible axis that does not surface a divergence. Defaults to one cent, matching the registry’s standard rounding tolerance. The operator-count axis is an exact integer match with no tolerance.

Return type:

Modelo349OperadorTotalsParity

Returns:

A Modelo349OperadorTotalsParity verdict. is_consistent is False whenever the reconstructed operator count differs at all, or the reconstructed base imponible diverges from base_summary_total by more than tolerance — a dropped or double-counted operador row must surface as a divergence, never silently collapse into is_consistent=True.