aeat.domain.calculations.registry._counterpart_bindings module

Counterpart-source registry binding helpers.

class CounterpartAggregationObservation(**data)[source]

Bases: BaseModel

One factual line from the user’s counterpart aggregation source.

Mirrors InvoiceObservation plus a source_kind field that is matched against the declared counterpart-source binding.

Parameters:

  • source_kind (CounterpartSourceKind)

  • source_id (str)

  • party_tax_id (str)

  • country_code (str)

  • transaction_date (date)

  • base_amount (Decimal)

  • invoice_total_amount (Decimal | 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)

source_kind: CounterpartSourceKind
source_id: str
party_tax_id: str
country_code: str
transaction_date: date
base_amount: Decimal
invoice_total_amount: Decimal | None
intracommunity_clave: str | None
is_rectification: bool
rectified_year: int | None
rectified_period: str | None
rectified_base_previous: Decimal | None
class CounterpartObservationRequirement(**data)[source]

Bases: BaseModel

Counterpart slice declared by one or more counterpart-source bindings.

Parameters:

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

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

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

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

binding_ids: tuple[BindingId, ...]
source_kinds: tuple[str, ...]
claves: tuple[str, ...]
rectification_scope: _RectificationScope
validate_counterpart_binding(binding)[source]

Validate a counterpart-source binding at registry-build time.

Accumulating list[str] validator: validates the selector against _InvoiceSelector and lifts the counterpart fact/op invariants to build time, preserving the underlying pydantic field error.

Return type:

list[str]

Parameters:

binding (DataBindingDefinition)

counterpart_binding_requirements(revision)[source]

Return CounterpartObservationRequirement slices needed by revision’s counterpart bindings.

Parameters:

revision (ModeloRevision) – The ModeloRevision whose counterpart bindings to inspect.

Return type:

tuple[CounterpartObservationRequirement, ...]

resolve_counterpart_binding_values(revision, observations)[source]

Resolve scalar counterpart-source bindings into Decimal aggregates.

Delegates to the shared invoice-family scalar resolver core (resolve_invoice_family_scalar_values()) parameterised by the counterpart membership set, the counterpart selector validator, and a source_kind-matched observation supplier.

Parameters:
Return type:

dict[TypeAliasType, Decimal]

resolve_counterpart_binding_row_values(revision, observations)[source]

Resolve row-producer counterpart-source bindings into per-row indexed values.

Delegates to the shared invoice-family row resolver core (resolve_invoice_family_row_values()) with cohort_by_source = True so a different counterpart source kind does not share row indexes.

Parameters:
Return type:

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