"""Strict immutable models for the invoice catalogue.
Defines the pydantic v2 records that back :mod:`cadrumo.domain.invoices`:
:class:`InvoiceLine`, :class:`Invoice`, and the keyed
:class:`InvoiceCatalogue`. Every model is strict, frozen, and forbids
extra fields; identity-bearing fields on :class:`Invoice` are
canonicalised in a ``model_validator`` and the stable
:attr:`Invoice.invoice_id` is derived via :func:`derive_invoice_id`.
Counterparty identity validation is delegated to
:mod:`cadrumo.domain.invoices.validators`.
"""
from __future__ import annotations
from collections.abc import Callable, Iterable, Iterator, Mapping
from datetime import date
from decimal import Decimal
from types import MappingProxyType
from typing import TYPE_CHECKING, Final, Self, cast, override
from pydantic import BaseModel, Field, field_serializer, field_validator, model_validator
from ...core.aggregation import IntracomOperationType, TravelAgencyMediationType
from ...core.country_code import CountryCodeAlpha2
from ...core.decimal.coercion import coerce_decimal
from ...core.errors.hierarchy import pydantic_validation_boundary
from ...core.external_constants import DEFAULT_CURRENCY
from ...core.hashing import content_hash_hex
from ...core.identity.bucket import BucketId
from ...core.identity.hex_ids import InvoiceId
from ...core.identity.tax_id import TaxIdIdentityToken
from ...core.models import STRICT_FROZEN_CONFIG as _STRICT_FROZEN
from ...core.money.rounding import CENT, round_to_cents
from ...core.time.utc import UtcInstant
from ...core.type_adapters import OBJECT_TUPLE_ADAPTER, STR_KEYED_MAPPING_ADAPTER
from ..calculations.registry.errors import RegistryValidationError
from ..calculations.registry.facts.resolution import MappingFactQuery, ResolvedMappingFact
from ..calculations.registry.governed_fact_scope import GovernedFactSource, governed_facts_in_scope
from ..calculations.registry.iva_category_catalogue import require_iva_category
from ..calculations.registry.iva_rate_kind_catalogue import require_iva_rate_kind
from ..calculations.registry.schema_base import DateAxis
from ..identifiers import canonical_decimal_string
from ..iva.classification import InvoiceKind, TransactionKind, resolve_transaction_kind_catalogue
from ..iva.errors import IvaRateNotFoundError, IvaValidationError
from ..iva.oss import OssIossRegime, resolve_oss_ioss_regime_catalogue
from ..iva.schema import EUMemberState, IvaCategory, IvaRateKind, require_eu_member_state, spanish_eu_member_state
from . import normalization as _normalization
from ._payload_normalisation import normalise_invoice_enum_fields, normalise_invoice_string_fields
from .enums import (
InvoiceClass,
InvoiceLegalMention,
InvoiceOperationDateRole,
IvaRate,
PaymentStatus,
advance_payment_received_role,
default_invoice_class,
invoice_class_rectificativa,
invoice_class_simplificada,
iva_rate_percentage,
resolve_iva_rate_token,
)
from .errors import InvoiceValidationError
if TYPE_CHECKING:
pass
from ...core.parsing.codes import IsoCurrencyCode
from ...core.time.clock import today_madrid
from .validators import (
is_eu_member_state_code,
)
"""Rounding slack allowed between a declared retención amount and rate.
The invoice-level totals are compared *exactly* because each is a sum of
line figures that were themselves already rounded to the cent. A retención
amount is not a sum: it is a rate applied to the base, so the recorded figure
legitimately differs from the recomputed product in the last cent depending on
where the issuer rounded. One cent is the same slack
:data:`~core.money.rounding.CENT` grants the line-level ``subtotal * iva_rate``
product, for the same reason.
"""
_SIMPLIFICADA_MANDATORY_TAX_ID_FACT_ID = "invoice-simplificada-counterparty-tax-id-applicability"
def _simplificada_mandatory_tax_id_categories(
*,
authority: GovernedFactSource | None = None,
) -> frozenset[IvaCategory]:
"""Resolve the dated simplified-invoice applicability catalogue."""
selected_authority = authority or governed_facts_in_scope()
if selected_authority is None:
raise RegistryValidationError(
"invoice simplified-tax-ID applicability requires an explicit authority operation or scope",
)
resolved = selected_authority.resolve_governed_fact(
MappingFactQuery(
fact_id=_SIMPLIFICADA_MANDATORY_TAX_ID_FACT_ID,
date_axis=DateAxis.FILING_PERIOD,
effective_date=today_madrid(),
),
)
if not isinstance(resolved, ResolvedMappingFact):
raise RegistryValidationError("invoice applicability must resolve as a mapping fact")
entries = {str(entry.key): str(entry.value) for entry in resolved.payload.entries}
try:
encoded = entries["mandatory_tax_id_categories"]
except KeyError as exc:
raise RegistryValidationError("invoice applicability is missing mandatory_tax_id_categories") from exc
try:
return frozenset(
require_iva_category(token.strip(), authority=selected_authority)
for token in encoded.split(",")
if token.strip()
)
except ValueError as exc:
raise RegistryValidationError("invoice applicability contains an unknown IVA category") from exc
_COLLECTED_PAYMENT_STATUSES: Final[frozenset[PaymentStatus]] = frozenset(
{PaymentStatus.PAID, PaymentStatus.PARTIALLY_PAID},
)
"""Payment states consistent with LIVA art. 75.Dos's "cobro total o parcial"."""
[docs]
def derive_invoice_id(
*,
kind: InvoiceKind,
invoice_number: str,
issued_at: date,
counterparty_tax_id: str | None,
currency: str,
grand_total: Decimal,
) -> str:
"""Return the stable invoice hash for one invoice record.
The digest is computed over a canonical JSON payload so that two
invoices with equal logical identity produce identical IDs regardless
of whitespace or numeric formatting.
Args:
kind: Invoice direction (issued / received).
invoice_number: AEAT-significant invoice number as printed on the
document.
issued_at: ISO calendar date printed on the invoice.
counterparty_tax_id: Counterparty NIF / NIE / CIF / IVA number
already validated and uppercased, or ``None`` for a factura
simplificada issued outside the RD 1619/2012 art. 6.1.d cases
that make the counterparty's tax id mandatory.
currency: ISO-4217 currency code already uppercased.
grand_total: Invoice grand total.
Returns:
A lowercase SHA-256 digest that uniquely identifies the invoice.
"""
return content_hash_hex(
{
"counterparty_tax_id": counterparty_tax_id or "",
"currency": currency,
"grand_total": canonical_decimal_string(grand_total),
"invoice_number": invoice_number,
"issued_at": issued_at.isoformat(),
"kind": kind.value,
}
)
[docs]
class InvoiceLine(BaseModel):
"""Immutable line item on an invoice."""
model_config = _STRICT_FROZEN
description: str = Field(min_length=1)
quantity: Decimal
unit_price: Decimal
subtotal: Decimal
iva_rate: IvaRate
iva_amount: Decimal
# Named for the taxonomy it belongs to, not "category" bare. This aggregate
# already carries `Invoice.iva_category`, a completely unrelated axis: one
# is the IVA TREATMENT of the operation, the other a SPENDING classification
# of the line. Two fields called "category" on one aggregate is how a reader
# reaches for the wrong one, and how a grep for either finds both.
#
# Currently written and persisted but read by no production consumer. Kept
# rather than removed because per-line spending classification is a real
# capability the aggregate is shaped for; the honest state is recorded here
# so a reader does not infer from its presence that aggregation consumes it.
spending_category_id: str | None = None
oss_rate_kind: IvaRateKind | None = None
@model_validator(mode="before")
@classmethod
@pydantic_validation_boundary
def _coerce_inputs(cls, data: object) -> object:
"""Coerce JSON-decoded strings into their strict pydantic types."""
if isinstance(data, cls):
return data
if not isinstance(data, Mapping):
return data
payload = STR_KEYED_MAPPING_ADAPTER.validate_python(data)
for key in ("quantity", "unit_price", "subtotal", "iva_amount"):
if key in payload:
payload[key] = coerce_decimal(payload[key])
if "iva_rate" in payload and isinstance(payload["iva_rate"], str):
payload["iva_rate"] = resolve_iva_rate_token(payload["iva_rate"], today_madrid())
if "oss_rate_kind" in payload and isinstance(payload["oss_rate_kind"], str):
stripped = payload["oss_rate_kind"].strip()
payload["oss_rate_kind"] = (
require_iva_rate_kind(stripped, effective_date=today_madrid()) if stripped else None
)
return payload
@field_validator("oss_rate_kind")
@classmethod
@pydantic_validation_boundary
def _validate_oss_rate_kind(cls, value: IvaRateKind | None) -> IvaRateKind | None:
"""Refuse an OSS/IOSS rate tier absent from the IVA facts catalogue."""
if value is None:
return None
return require_iva_rate_kind(value, effective_date=today_madrid())
@field_validator("description")
@classmethod
@pydantic_validation_boundary
def _trim_description(cls, value: str) -> str:
trimmed = value.strip()
if not trimmed:
raise InvoiceValidationError("description must not be blank")
return trimmed
@field_validator("spending_category_id")
@classmethod
@pydantic_validation_boundary
def _validate_spending_category_id(cls, value: str | None) -> str | None:
if value is None:
return None
trimmed = value.strip()
if not trimmed:
raise InvoiceValidationError("spending_category_id must not be blank")
return trimmed
@field_validator("quantity")
@classmethod
@pydantic_validation_boundary
def _require_positive_quantity(cls, value: Decimal) -> Decimal:
if value <= Decimal("0"):
raise InvoiceValidationError("quantity must be strictly positive")
return value
@field_validator("unit_price", "subtotal", "iva_amount")
@classmethod
@pydantic_validation_boundary
def _require_non_negative(cls, value: Decimal) -> Decimal:
if value < Decimal("0"):
raise InvoiceValidationError("monetary value must be non-negative")
return value
@model_validator(mode="after")
@pydantic_validation_boundary
def _validate_arithmetic(self) -> Self:
expected_subtotal = (self.quantity * self.unit_price).quantize(Decimal("0.0001"))
if abs(self.subtotal - expected_subtotal) > CENT:
raise InvoiceValidationError("subtotal must equal quantity * unit_price within 1 cent")
# IVA-rate arithmetic has an explicit devengo-date dependency. A line
# deliberately has no date of its own, so the enclosing Invoice checks
# it once its operation/issue date is available rather than parsing a
# number from this persisted token.
return self
[docs]
class Invoice(BaseModel):
"""Strict frozen record for one issued or received invoice."""
model_config = _STRICT_FROZEN
invoice_id: InvoiceId
bucket_id: BucketId | None = Field(default=None)
kind: InvoiceKind
invoice_class: InvoiceClass = Field(default_factory=default_invoice_class)
series: str | None = Field(default=None, min_length=1)
invoice_number: str = Field(min_length=1)
issued_at: date
operation_date: date | None = None
operation_date_role: InvoiceOperationDateRole | None = None
counterparty_name: str = Field(min_length=1)
counterparty_tax_id: TaxIdIdentityToken | None = None
counterparty_country: CountryCodeAlpha2
# Which Member State IVA-IDENTIFIES the counterparty, read from the prefix
# of the IVA number the document printed. A DIFFERENT fact from
# `counterparty_country` above, which is an address -- establishment -- and
# never a source for this one: a Spanish-established acquirer can hold a
# German IVA number, and a German-established one can purchase under a
# Spanish NIF-IVA. Ley 37/1992 art. 25 exempts on this fact, not on the
# address, so deriving one from the other lands in money in both
# directions. `None` means the identification was not established -- never
# that the party is identified nowhere, and above all never that it is
# identified in Spain.
counterparty_identification_state: EUMemberState | None = None
# RD 1619/2012 art. 6.1.e: "Domicilio, tanto del obligado a expedir
# factura como del destinatario de las operaciones." Named by legal role,
# not by party-relative-to-us like `counterparty_*` above, because the
# two roles swap sides with `kind`: for an ISSUED invoice the issuer is
# self and the recipient is the counterparty; for a RECEIVED invoice the
# issuer is the counterparty and the recipient is self. Each field is
# what that document actually PRINTED -- never derived from a current
# profile domicilio, which may differ from what an older invoice states.
issuer_address: str | None = Field(default=None, min_length=1)
recipient_address: str | None = Field(default=None, min_length=1)
# RD 1619/2012 art. 6.1.j / .l / .m / .n / .o / .p: the fixed legal
# notices ("menciones") the reglamento requires printed under specific
# regimes. `exemption_reference` is art. 6.1.j's REFERENCE (a Directiva
# 2006/112/CE provision, a LIVA article, or a bare "operación exenta"
# statement) -- free text because the reglamento does not fix its
# wording. `legal_mentions` carries the CLOSED, literally-quoted phrases
# of art. 6.1.l/.m/.n/.o/.p (see `InvoiceLegalMention`). Both are
# evidence of what the issuer printed; neither is derived from
# `iva_category` below -- that would manufacture evidence of compliance
# nobody observed on the document.
exemption_reference: str | None = Field(default=None, min_length=1)
legal_mentions: tuple[InvoiceLegalMention, ...] = ()
base_total: Decimal
iva_total: Decimal
grand_total: Decimal
currency: IsoCurrencyCode
lines: tuple[InvoiceLine, ...]
payment_status: PaymentStatus
linked_transaction_ids: tuple[str, ...] = ()
notes: str = ""
iva_category: IvaCategory | None = None
operation_type: IntracomOperationType | None = None
# RD 1619/2012 disposición adicional cuarta: set when this invoice
# documents a travel-agency mediation service (see
# :class:`~core.aggregation.TravelAgencyMediationType`). Feeds Modelo 347 claves F/G
# -- never inferred from `iva_category`, which carries no mediation axis.
travel_agency_mediation: TravelAgencyMediationType | None = None
# RD 1065/2007 art. 34.g: when the filer collects this amount on behalf
# of a socio, asociado or colegiado (Modelo 347 clave C), the declared
# counterparty is the BENEFICIARY whose fees were collected, not
# necessarily whoever paid this invoice. `None` means this invoice is
# NOT a third-party collection -- the ordinary counterparty declares.
collected_on_behalf_of_tax_id: TaxIdIdentityToken | None = None
collected_on_behalf_of_name: str | None = Field(default=None, min_length=1)
# RD 1065/2007 art. 33.1: "las subvenciones, auxilios o ayudas no
# reintegrables que puedan otorgar o recibir" are declarable operations
# distinct from ordinary payments. Feeds Modelo 347 clave E, which is
# additionally exclusive to a PUBLIC_ADMINISTRATION_ENTITY filer
# (art. 31.2's second paragraph) -- this fact alone never classifies E.
# Tri-state on purpose: ``None`` is UNDECLARED, distinct from ``False``.
# A silent ``False`` default would under-declare clave E for every
# subvención nobody thought to flag; the resolver surfaces an advisory
# for a public-administration filer's invoice left ``None`` rather than
# picking a side.
is_subvencion_ayuda: bool | None = None
# RD 1065/2007 art. 31.1's last paragraph and art. 31.2: an acquisition
# made "al margen de las actividades empresariales o profesionales" by a
# filer carrying one of clave D's four roles. A mixed-activity filer (a
# public university with a cafeteria concession, say) has SOME
# acquisitions inside its own economic activity and some outside it, so
# this is a per-invoice fact, never inferable from the filer's role
# alone. Tri-state for the same reason as ``is_subvencion_ayuda``: a
# comunidad de propietarios earning rooftop-antenna income has genuine
# business-adjacent acquisitions too, so neither a blanket ``True`` nor
# ``False`` default is safe for its population either.
outside_economic_activity: bool | None = None
oss_ioss_regime: OssIossRegime | None = None
oss_transaction_kind: TransactionKind | None = None
retention_rate: Decimal | None = None
retention_amount: Decimal | None = None
recargo_amount: Decimal | None = None
suplido_amount: Decimal | None = None
rectifies_invoice_number: str | None = Field(default=None, min_length=1)
payment_id: str | None = None
fx_rate: Decimal | None = None
fx_rate_date: date | None = None
# WHO quoted the rate, at parity with the ledger transaction's own
# `rate_source`. The rate and its date said what was applied and when, but
# not on whose published series, so a stored euro total on a foreign invoice
# could not be re-derived or challenged years later -- while the very same
# conversion recorded on a bank row could. Set together with the rate, for
# the same reason the rate and its date are: half a provenance is a claim
# nothing can check.
fx_rate_source: str | None = Field(default=None, min_length=1)
# When this RECORD was entered and last amended, which is a different fact
# from `issued_at` (when the document was issued) and from `operation_date`
# (when the operation occurred). Both are outside the identity derived by
# `derive_invoice_id`: folding a clock into the id would mint a new record
# on every retry, which `aeat-cli-contract`
# bars. They are last-seen body fields, not identity.
#
# Optional rather than required on purpose. A record that genuinely carries
# no recorded entry time must say so, because the alternative is stamping
# `now()` at load and manufacturing an audit fact nobody observed. `None`
# here means "not recorded", never "recorded as now".
created_at: UtcInstant | None = None
updated_at: UtcInstant | None = None
@override
def __hash__(self) -> int:
return hash(self.invoice_id)
@property
def base_total_eur(self) -> Decimal | None:
"""``base_total`` in euro, or ``None`` when the invoice is unconverted."""
return self._in_eur(self.base_total)
@property
def iva_total_eur(self) -> Decimal | None:
"""``iva_total`` in euro, or ``None`` when the invoice is unconverted."""
return self._in_eur(self.iva_total)
@property
def grand_total_eur(self) -> Decimal | None:
"""``grand_total`` in euro, or ``None`` when the invoice is unconverted."""
return self._in_eur(self.grand_total)
@property
def retention_amount_eur(self) -> Decimal | None:
"""The declared retención in euro, or ``None`` when there is none to convert.
Returns ``None`` both when no retención was declared and when the
invoice is unconverted, so a caller that needs to tell those apart must
read :attr:`retention_amount` alongside it -- the same shape the three
total accessors already have.
"""
if self.retention_amount is None:
return None
return self._in_eur(self.retention_amount)
@property
def recargo_amount_eur(self) -> Decimal | None:
"""The recargo de equivalencia in euro, or ``None`` when there is none to convert.
Same shape as :attr:`retention_amount_eur`: ``None`` covers both "no
recargo was declared" and "the invoice is unconverted", so a caller
needing to tell those apart reads :attr:`recargo_amount` alongside it.
"""
if self.recargo_amount is None:
return None
return self._in_eur(self.recargo_amount)
@property
def suplido_amount_eur(self) -> Decimal | None:
"""The suplido (LIVA art. 78.Tres.3.º) in euro, or ``None`` when there is none to convert.
Same shape as :attr:`recargo_amount_eur` and :attr:`retention_amount_eur`:
``None`` covers both "no suplido was declared" and "the invoice is
unconverted".
"""
if self.suplido_amount is None:
return None
return self._in_eur(self.suplido_amount)
def _in_eur(self, amount: Decimal) -> Decimal | None:
"""Convert *amount* to euro using the stored rate.
A euro invoice is already euro and converts to itself. A foreign
invoice converts only when :attr:`fx_rate` was resolved at ingest;
without it the euro value is genuinely unknown and is reported as
``None`` so the caller gates the invoice, rather than returning the
face value and silently declaring foreign units as euro.
"""
if self.currency == DEFAULT_CURRENCY:
return amount
if self.fx_rate is None:
return None
return round_to_cents(amount * self.fx_rate)
[docs]
def line_amount_eur(self, amount: Decimal) -> Decimal | None:
"""Convert a LINE-level native-currency amount to euro, or ``None`` if unconverted.
:class:`InvoiceLine` carries no currency of its own -- every line on
an invoice is denominated in that invoice's own :attr:`currency`
(a single document states one currency for all its lines), so a line
amount (``line.subtotal``, ``line.iva_amount``) converts through the
SAME rate resolution as the invoice-level totals
(:attr:`base_total_eur` and siblings), not a separate per-line
mechanism. Delegating to :meth:`_in_eur` keeps that arithmetic
consistent: summing ``line_amount_eur(line.subtotal)`` over every line
equals :attr:`base_total_eur` exactly, the same way summing
``line.subtotal`` natively already equals :attr:`base_total`.
Args:
amount: A native-currency line amount, e.g. ``line.subtotal`` or
``line.iva_amount`` for one of this invoice's own
:attr:`lines`.
Returns:
The euro-equivalent amount, or ``None`` when this invoice is
foreign-currency with no resolved :attr:`fx_rate`.
"""
return self._in_eur(amount)
@model_validator(mode="before")
@classmethod
@pydantic_validation_boundary
def _normalise_and_derive_invoice_id(cls, data: object) -> object:
"""Canonicalise identity-bearing fields and derive ``invoice_id``."""
if isinstance(data, cls):
return data
if not isinstance(data, Mapping):
return data
payload = STR_KEYED_MAPPING_ADAPTER.validate_python(data)
# Typed explicitly: the tuple mixes named normalisers with two lambdas,
# and without a declared element type the lambda parameters and the
# loop variable all read as unknown.
normalisers: tuple[Callable[[dict[str, object]], dict[str, object]], ...] = (
_normalization.normalise_invoice_dates,
normalise_invoice_enum_fields,
normalise_invoice_string_fields,
_normalization.normalise_invoice_counterparty,
_normalization.normalise_invoice_currency,
_normalization.normalise_invoice_monetary_fields,
lambda payload: _normalization.derive_invoice_id_when_complete(
payload,
derive_invoice_id=derive_invoice_id,
),
lambda payload: _normalization.normalise_invoice_collections(
payload,
normalise_linked_transaction_ids=_normalise_linked_transaction_ids,
),
_normalization.normalise_invoice_payment_id,
)
for normalise in normalisers:
payload = normalise(payload)
return payload
@field_validator("base_total", "iva_total", "grand_total")
@classmethod
@pydantic_validation_boundary
def _require_non_negative_totals(cls, value: Decimal) -> Decimal:
if value < Decimal("0"):
raise InvoiceValidationError("invoice totals must be non-negative")
return value
@field_validator("lines")
@classmethod
@pydantic_validation_boundary
def _require_lines(cls, value: tuple[InvoiceLine, ...]) -> tuple[InvoiceLine, ...]:
if not value:
raise InvoiceValidationError("invoice must carry at least one line")
return value
@model_validator(mode="after")
@pydantic_validation_boundary
def _validate_fx_conversion_coherence(self) -> Self:
"""Reject an incoherent conversion stamp.
The triple is all-or-nothing so a stored rate is always auditable: a
rate without its date cannot be located in a published series, a rate
without its source does not say whose series to look in, and a date or
source without a rate converts nothing. A euro invoice carries none of
them -- a stamp there would imply a conversion that never happened.
"""
stamp_flags = (
self.fx_rate is not None,
self.fx_rate_date is not None,
self.fx_rate_source is not None,
)
stamp_present = any(stamp_flags)
fx_rate = self.fx_rate if self.fx_rate is not None else Decimal("0")
_normalization.raise_first_invoice_violation(
(
(
(self.currency == DEFAULT_CURRENCY, stamp_present) == (True, True),
"a EUR invoice must not carry an fx conversion stamp",
),
(
stamp_flags not in ((False, False, False), (True, True, True)),
"fx_rate, fx_rate_date and fx_rate_source must be set together",
),
(
(stamp_flags[0], fx_rate <= Decimal("0")) == (True, True),
"fx_rate must be strictly positive",
),
),
)
return self
@model_validator(mode="after")
@pydantic_validation_boundary
def _validate_line_rates_were_in_force(self) -> Self:
"""Refuse a line naming a rate the statute did not offer on the devengo date.
The invoice is the first object that knows WHEN the operation happened,
so it is the first that can ask whether each line's rate was legally
available. The question is live because the taxonomy carries the RD-ley
4/2024 food rates: 2 % and 4 % were both correct super-reducido rates in
late 2024, so a slot cannot be checked against its tier alone, and a
2025 invoice claiming 2 % is naming a rate already withdrawn.
Checked against the operation date when one is recorded and the issue
date otherwise -- RD 1619/2012 art. 6.1.i records the operation date
precisely when it differs from expedition, and both readings are the
LIVA art. 75 devengo date the rate binds to (art. 90.Dos).
"""
devengo_date = self.operation_date or self.issued_at
for line in self.lines:
try:
rate = iva_rate_percentage(line.iva_rate, devengo_date)
except IvaRateNotFoundError as exc:
raise InvoiceValidationError(
f"line rate {line.iva_rate.name} was not in force on {devengo_date.isoformat()}: {exc}",
) from exc
if line.oss_rate_kind is not None:
continue
if rate is None:
if line.iva_amount != Decimal("0"):
raise InvoiceValidationError("iva_amount must be zero for nonnumeric IVA lines")
continue
expected_iva = (line.subtotal * rate).quantize(Decimal("0.0001"))
if abs(line.iva_amount - expected_iva) > CENT:
raise InvoiceValidationError("iva_amount must equal subtotal * iva_rate within 1 cent")
return self
@model_validator(mode="after")
@pydantic_validation_boundary
def _validate_totals_and_exempt_invariants(self) -> Self:
line_subtotal_sum = sum((line.subtotal for line in self.lines), start=Decimal("0"))
line_iva_sum = sum((line.iva_amount for line in self.lines), start=Decimal("0"))
_normalization.require_equal(
self.base_total,
line_subtotal_sum,
"base_total must equal the exact sum of line subtotals",
)
_normalization.require_equal(
self.iva_total,
line_iva_sum,
"iva_total must equal the exact sum of line iva amounts",
)
recargo = self.recargo_amount or Decimal("0")
suplido = self.suplido_amount or Decimal("0")
_normalization.require_equal(
self.grand_total,
self.base_total + self.iva_total + recargo + suplido,
"grand_total must equal base_total + iva_total + recargo_amount + suplido_amount exactly",
)
devengo_date = self.operation_date or self.issued_at
all_non_numeric = all(iva_rate_percentage(line.iva_rate, devengo_date) is None for line in self.lines)
if all_non_numeric:
# Checked before the grand-total equality below so the operator is
# told which component is impossible, rather than being handed a
# totals mismatch they would have to decompose themselves. The
# recargo rides on the cuota of a taxable supply (LIVA art. 161),
# so a supply bearing no cuota by law bears no recargo either. A
# suplido is unrelated to whether the underlying supply is taxable
# -- it is a disbursement made in the client's name (LIVA
# art. 78.Tres.3.º) -- so it stays permitted here.
_normalization.raise_first_invoice_violation(
(
(
self.iva_total != Decimal("0"),
"iva_total must be zero when every line is nonnumeric",
),
(
recargo != Decimal("0"),
"recargo_amount must be zero when every line is nonnumeric",
),
(
self.grand_total != self.base_total + suplido,
"grand_total must equal base_total + suplido_amount when every line is nonnumeric",
),
),
)
return self
@model_validator(mode="after")
@pydantic_validation_boundary
def _validate_retencion_consistency(self) -> Self:
"""Enforce the retención invariants, holding retención outside the totals.
Retención is an IRPF settlement-side deduction, not a price component.
The canonical per-invoice identity is
``total (contraprestación) = base_total + iva_total + recargo_amount
+ suplido_amount`` and, separately, ``cash = total - retención``: the withholding changes
what the payer *transfers*, never what the operation *costs*. So
:attr:`grand_total` stays retención-inclusive, and an issuer who nets
the withholding out of the grand total is refused by
:meth:`_validate_totals_and_exempt_invariants` above, whose equality is
exact. Nothing here re-checks that; this validator governs only the two
retención fields themselves. Recargo sits on the opposite side of that
identity for the reason :meth:`_validate_recargo_consistency` gives.
The retención base is the **base imponible**, not the grand total: RIRPF
art. 95.1 withholds "sobre los ingresos íntegros satisfechos", and the
IVA repercutido is not an ingreso of the issuer (PGC NRV 12.ª/14.ª). A
rate checked against :attr:`grand_total` would therefore over-state the
expected withholding by the whole cuota.
:attr:`retention_rate` is a **fraction**, matching
:func:`~cadrumo.domain.invoices.enums.iva_rate_percentage` and the registry
RIRPF art. 95 rates, both of which express a rate as ``pct / 100``. The
upper bound is what catches a percentage written into a fractional
field: ``15`` for "15 %" is refused rather than silently read as
1500 %.
An amount may stand alone -- the invoice records what was withheld
without recording which rate produced it, which is how many issued
invoices actually read. A rate may not: on its own it declares a
proportion of nothing, and inferring the amount from it would
manufacture a figure the document never stated.
"""
if self.retention_amount is not None and self.retention_amount < Decimal("0"):
raise InvoiceValidationError(
"retention_amount must be non-negative",
context={"fields": ("retention_amount",)},
)
if self.retention_rate is not None:
if self.retention_rate < Decimal("0") or self.retention_rate > Decimal("1"):
raise InvoiceValidationError(
"retention_rate must be a fraction between 0 and 1 (0.15 for a 15 % retención), not a percentage",
context={"fields": ("retention_rate",)},
)
if self.retention_amount is None:
raise InvoiceValidationError(
"retention_rate requires retention_amount; a rate alone declares no withheld figure",
context={"fields": ("retention_rate", "retention_amount")},
)
if self.retention_amount is not None and self.retention_amount > self.base_total:
raise InvoiceValidationError(
"retention_amount must not exceed base_total; the retención base is the "
"base imponible (ingresos íntegros), not the IVA-inclusive total",
context={"fields": ("retention_amount", "base_total")},
)
if self.retention_rate is not None and self.retention_amount is not None:
expected_retencion = (self.base_total * self.retention_rate).quantize(Decimal("0.0001"))
if abs(self.retention_amount - expected_retencion) > CENT:
raise InvoiceValidationError(
"retention_amount must equal base_total * retention_rate within 1 cent",
context={"fields": ("retention_rate", "retention_amount", "base_total")},
)
return self
@model_validator(mode="after")
@pydantic_validation_boundary
def _validate_recargo_consistency(self) -> Self:
"""Enforce the recargo de equivalencia invariants, holding it inside the total.
Recargo is the mirror image of retención and is deliberately modelled
as its opposite. Retención is settlement-side: it reduces what the
payer transfers without changing what the operation cost, so it sits
*outside* :attr:`grand_total`. Recargo is a price component: LIVA
art. 161 has the supplier repercutir it on the entrega alongside the
cuota, and the comerciante minorista genuinely owes it, so it sits
*inside* :attr:`grand_total`. Recording it outside would understate
what the customer was actually invoiced.
``None`` and ``Decimal("0")`` are both accepted and mean different
things. ``None`` is "this invoice makes no statement about recargo",
which is the ordinary case for the overwhelming majority of invoices
whose customer is not a comerciante minorista. An explicit zero is a
positive declaration that the régimen was considered and does not
apply. Neither is treated as evidence of the other, which is the whole
reason the field is nullable rather than defaulted to zero -- an
unrecorded surcharge must stay distinguishable from one that does not
arise.
The upper bound is :attr:`iva_total`. Every statutory recargo rate is
far below its companion IVA rate (5.2 % against 21 %, 1.4 % against
10 %, 0.5 % against 4 %), so a recargo exceeding the cuota it rides on
is arithmetically impossible under any tier and is far more likely to
be the cuota written into the wrong field. The bound is deliberately
loose rather than a per-tier rate check: no recargo rate table ships in
the registry, and inventing rate literals here would put regulatory
values in a feature module.
"""
if self.recargo_amount is None:
return self
_normalization.require_optional_non_negative(self.recargo_amount, "recargo_amount must be non-negative")
if self.recargo_amount > self.iva_total:
raise InvoiceValidationError(
"recargo_amount must not exceed iva_total; every recargo tier is a smaller "
"percentage than the IVA rate it accompanies (LIVA art. 161)",
)
return self
@model_validator(mode="after")
@pydantic_validation_boundary
def _validate_suplido_consistency(self) -> Self:
"""Enforce the suplido invariant, holding it inside the total alongside recargo.
A suplido (LIVA art. 78.Tres.3.º) is a sum paid by the issuer in the
client's name and on their behalf under an explicit mandate --
typically a third-party fee or disbursement passed through unchanged.
It is excluded from the base imponible by law, so it cannot join
:attr:`base_total` or :attr:`iva_total`; it is nonetheless something
the client owes the issuer, so -- like recargo -- it joins
:attr:`grand_total` and, through it, ``cash``. It takes a third
position on the identity rather than becoming a second recargo: unlike
recargo it carries no statutory rate and so no upper bound tied to
another figure on the invoice, and unlike recargo it is not restricted
to a taxable supply -- a suplido may accompany an exempt operation.
``None`` and ``Decimal("0")`` are both accepted and mean different
things, for the same reason :meth:`_validate_recargo_consistency`
keeps the two apart on recargo: ``None`` is "this invoice makes no
statement about a suplido"; an explicit zero is "one was considered
and none arose".
"""
if self.suplido_amount is None:
return self
_normalization.require_optional_non_negative(self.suplido_amount, "suplido_amount must be non-negative")
return self
@model_validator(mode="after")
@pydantic_validation_boundary
def _validate_invoice_class_consistency(self) -> Self:
"""Enforce the RD 1619/2012 art. 6.1 invoice-class axis.
A rectificativa is coupled to two facts art. 6.1 makes mandatory only
for that class: a specific issuing series (art. 6.1.a.2.º) and the
invoice it corrects (LIVA art. 89 requires a rectificativa to
rectify a named prior invoice). Neither field is meaningful outside
that class, so both are refused on any other one -- a stray
``rectifies_invoice_number`` on an ordinaria would silently assert a
correction the invoice's own class denies.
A factura simplificada's counterparty tax id is optional under art. 6
UNLESS one of the three art. 6.1.d cases this table can read from the
invoice's own data applies: an entrega intracomunitaria exenta (case
1.º) or an operation where the destinatario is the sujeto pasivo
(case 2.º, ``domestic_reverse_charge``). Case 3.º -- a domestic
operation where the issuer is established in the territorio de
aplicación del impuesto -- is NOT modelled here: this record carries
no field stating where its issuer is established, so it is a declared
gap rather than a guessed default. An ordinaria or rectificativa keeps
the tax id mandatory unconditionally, unchanged from before this
class axis existed.
The simplificada relief applies to the ISSUED side only.
:attr:`counterparty_tax_id` names a different party depending on
:attr:`kind`: on an ISSUED invoice it is the destinatario's NIF --
the fact art. 6.1.d's three cases actually govern -- but on a
RECEIVED invoice it is the ISSUER's OWN identification, which
art. 6.1.d's opening clause (and art. 7.1.d for a simplificada
specifically) keeps mandatory regardless of class. A received
simplificada with no supplier NIF is not a relieved case 3.º
scenario; it is a document missing its own issuer's identity, so
the relief must never fire for :attr:`kind` ``RECEIVED``.
An entrega intracomunitaria exenta is refused for the simplified class outright,
independent of the tax id question above: RD 1619/2012 art. 4.4.a)
forbids a factura simplificada for this category altogether, so a
document naming this category must never carry this class at all --
it is not merely a case where the tax id relief does not apply. Art. 4
also carries amount-based eligibility (the 400 EUR general and 3.000
EUR sector-specific ceilings) and a closed sector list neither this
record nor any field on it can express; that axis is a declared,
documented gap, not modelled here or anywhere on this class.
"""
category = self.iva_category
missing_tax_id = self.counterparty_tax_id is None
category_value = getattr(category, "value", "")
simplificada = invoice_class_simplificada()
rectificativa = invoice_class_rectificativa()
relief_case = (self.invoice_class, self.kind) == (simplificada, InvoiceKind.ISSUED)
_normalization.raise_first_invoice_violation(
(
(
(self.invoice_class == rectificativa, not self.series) == (True, True),
"a factura rectificativa must be issued in a specific series (RD 1619/2012 art. 6.1.a.2.º)",
),
(
(self.invoice_class == rectificativa, not self.rectifies_invoice_number) == (True, True),
"a factura rectificativa must name the invoice it rectifies (LIVA art. 89)",
),
(
(self.invoice_class != rectificativa, self.rectifies_invoice_number is not None) == (True, True),
"rectifies_invoice_number only applies to a factura rectificativa",
),
(
(self.invoice_class, category) == (simplificada, require_iva_category("intra_community_supply")),
"a factura simplificada must not be issued for an entrega intracomunitaria exenta "
"(RD 1619/2012 art. 4.4.a); issue an ordinaria or rectificativa instead",
),
(
(missing_tax_id, not relief_case) == (True, True),
"counterparty_tax_id is required unless the invoice is simplified and kind is ISSUED; "
"on a RECEIVED invoice it names the issuer's own identity, which stays mandatory",
),
(
(missing_tax_id, category in _simplificada_mandatory_tax_id_categories()) == (True, True),
"counterparty_tax_id is required on a factura simplificada whose iva_category is "
f"{category_value!r} (RD 1619/2012 art. 6.1.d)",
),
),
)
return self
@model_validator(mode="after")
@pydantic_validation_boundary
def _validate_intracommunity_acquirer_identification(self) -> Self:
"""Refuse an entrega intracomunitaria exenta to a Spanish-IDENTIFIED acquirer.
RD 1619/2012 art. 6.1.d requires the invoice state "el Número de
Identificación Fiscal ... atribuido por la Administración ... de otro
Estado miembro", and LIVA art. 25 exempts on that identification. A
counterparty purchasing under a Spanish IVA identification therefore
contradicts the declared category outright, and the contradiction is
what this refuses.
It reads the identification and NOT
:attr:`counterparty_country`, which is an address. The two diverge in
real trade, and keying this on the address refused a supply art. 25
exempts: a Spanish-established acquirer holding a French IVA number is
an intra-community acquirer, and could not previously be recorded at
all. Establishment is a question about where a party IS, and it does
not answer this one.
ABSENT identification is NOT refused here. It is not a contradiction --
it is a fact not yet recorded, and inferring it from the address is the
substitution this whole field exists to remove. The aggregation gate
withholds such a row with a resolvable review item naming the fact to
supply, which is where an unanswered question belongs.
Deliberately not an EU-membership check: Northern Ireland (``XI``) and
a non-EU destination are both legitimate declared facts a later
M349/export classification decides between; this guard only refuses the
one identification the category can never legitimately name.
"""
if (
self.iva_category == require_iva_category("intra_community_supply")
and self.counterparty_identification_state == spanish_eu_member_state()
):
raise InvoiceValidationError(
"an entrega intracomunitaria exenta cannot name an acquirer purchasing under a "
"Spanish IVA identification (LIVA art. 25); its country of establishment does not "
"change that",
)
return self
@model_validator(mode="after")
@pydantic_validation_boundary
def _validate_operation_date_consistency(self) -> Self:
"""Enforce the LIVA art. 75 devengo-date axis art. 6.1.i lets an invoice state.
``operation_date`` and ``operation_date_role`` travel together: a date
with no stated role would leave a reader guessing which of art. 6.1.i's
two clauses it answers, and a role with no date states nothing.
A pago anticipado devengo (the advance-payment role, LIVA
art. 75.Dos) requires money to have actually been received -- "el
cobro total o parcial del precio" -- so it is refused against a
``payment_status`` that states none was. It is also refused outright
on ``INTRA_COMMUNITY_SUPPLY``: art. 75.Dos, párrafo segundo, excludes
"las entregas de bienes comprendidas en el artículo 25" from the
pagos-anticipados rule, so that category always devengues under
art. 75.Uno.8.º regardless of any advance received.
"""
if (self.operation_date is None) != (self.operation_date_role is None):
raise InvoiceValidationError("operation_date and operation_date_role must be set together")
advance_payment_role = advance_payment_received_role()
if self.operation_date_role == advance_payment_role:
if self.iva_category == require_iva_category("intra_community_supply"):
raise InvoiceValidationError(
"a pago anticipado devengo does not apply to an entrega intracomunitaria exenta "
"(LIVA art. 75.Dos, párrafo segundo, excludes art. 25 entregas)",
)
if self.payment_status not in _COLLECTED_PAYMENT_STATUSES:
raise InvoiceValidationError(
f"operation_date_role {advance_payment_role.value} requires a collected payment_status "
"(PAID or PARTIALLY_PAID); LIVA art. 75.Dos devengues on actual cobro",
)
return self
@model_validator(mode="after")
@pydantic_validation_boundary
def _validate_oss_ioss_axes(self) -> Self:
"""Validate the optional OSS/IOSS projection axes used by Modelo 369."""
has_oss_line_rate = any(line.oss_rate_kind is not None for line in self.lines)
axis_flags = (self.oss_ioss_regime is None, self.oss_transaction_kind is None)
if axis_flags != (False, False):
_normalization.raise_first_invoice_violation(
(
(
(axis_flags, has_oss_line_rate) == ((True, True), True),
"oss_rate_kind requires invoice-level OSS/IOSS axes",
),
(
axis_flags not in ((True, True), (False, False)),
"oss_ioss_regime and oss_transaction_kind must be supplied together",
),
),
)
return self
operation = governed_facts_in_scope()
from ..calculations.registry.authority import PinnedAuthorityOperation
if not isinstance(operation, PinnedAuthorityOperation):
raise RegistryValidationError(
"invoice OSS/IOSS validation requires a generation-pinned authority operation",
)
try:
transaction_kind = resolve_transaction_kind_catalogue(
self.operation_date or self.issued_at,
operation=operation,
).require(self.oss_transaction_kind)
except IvaValidationError as exc:
raise InvoiceValidationError("oss_transaction_kind must be declared by the facts registry") from exc
try:
catalogue = resolve_oss_ioss_regime_catalogue(
effective_date=self.operation_date or self.issued_at,
authority=operation,
)
regime = catalogue.require(self.oss_ioss_regime)
except RegistryValidationError as exc:
raise InvoiceValidationError("oss_ioss_regime must be declared by the facts registry") from exc
_normalization.raise_first_invoice_violation(
(
(
self.kind is not InvoiceKind.ISSUED,
"OSS/IOSS invoice projection only applies to issued invoices",
),
(
self.counterparty_eu_member_state is None,
"OSS/IOSS invoice projection requires an EU destination member state",
),
(
transaction_kind.value not in catalogue.transaction_kinds_for(regime),
"oss_transaction_kind is not valid for the supplied oss_ioss_regime",
),
),
)
return self
@property
def counterparty_eu_member_state(self) -> EUMemberState | None:
"""Return the substrate-typed EUMemberState for the counterparty, or ``None`` for non-EU.
:attr:`counterparty_country` carries the raw uppercase ISO-3166-1
alpha-2 code (validated at construction time). This typed
accessor lets downstream consumers (Modelo 369 OSS bindings,
intra-community classification, OSS classifier dispatch) work with a
registry-projected token without a per-call lowercase / membership
check. It is anchored to the dated EU member-state fact 0131.
"""
if not is_eu_member_state_code(self.counterparty_country):
return None
return require_eu_member_state(self.counterparty_country)
def _normalise_linked_transaction_ids(value: object) -> tuple[str, ...]:
"""Deduplicate-preserve-order and validate the shape of linked transaction IDs."""
_normalization.raise_first_invoice_violation(
(
(
isinstance(value, str | bytes),
"linked_transaction_ids must be a sequence of IDs, not a single string",
),
(not isinstance(value, Iterable), "linked_transaction_ids must be iterable"),
),
)
seen: dict[str, None] = {}
for item in OBJECT_TUPLE_ADAPTER.validate_python(value):
_normalization.raise_first_invoice_violation(
((not isinstance(item, str), "each linked_transaction_id must be a string"),),
)
normalized = cast(str, item).strip().lower()
_normalization.raise_first_invoice_violation(
(
(
not _is_hex_digest(normalized, length=64),
"each linked_transaction_id must be a 64-character lowercase hex digest",
),
),
)
seen.setdefault(normalized, None)
return tuple(seen.keys())
def _is_hex_digest(value: str, *, length: int) -> bool:
return len(value) == length and all(char in "0123456789abcdef" for char in value)
[docs]
class InvoiceCatalogue(BaseModel):
"""Immutable invoice catalogue keyed by ``invoice_id``."""
model_config = _STRICT_FROZEN
invoices: Mapping[str, Invoice] = Field(default_factory=dict)
@model_validator(mode="before")
@classmethod
@pydantic_validation_boundary
def _coerce_catalogue_input(cls, data: object) -> object:
"""Accept a catalogue, its canonical payload, or an iterable of invoices.
A mapping arrives here as one of two things: the canonical serialized
payload, which carries its entries under ``invoices``, or the field
kwargs of a direct construction, which for a catalogue with no entries
is empty. A NON-EMPTY mapping with no ``invoices`` key is neither. It
is a catalogue serialized without its wrapper -- a shape nothing in
this codebase writes -- and it is refused rather than wrapped: wrapping
promotes an arbitrary mapping into a catalogue whose keys no writer
ever established to be invoice ids, and the resulting record is
indistinguishable afterwards from one that was written correctly.
"""
if isinstance(data, cls):
return data
if isinstance(data, Mapping):
payload = STR_KEYED_MAPPING_ADAPTER.validate_python(data)
if payload and "invoices" not in payload:
raise InvoiceValidationError(
"invoice catalogue payload must carry its entries under the 'invoices' key; "
f"got a bare mapping of {len(payload)} top-level entries",
)
return payload
if isinstance(data, Iterable) and not isinstance(data, str | bytes):
invoices: dict[str, Invoice] = {}
for item in OBJECT_TUPLE_ADAPTER.validate_python(data):
invoice = item if isinstance(item, Invoice) else Invoice.model_validate(item)
if invoice.invoice_id in invoices:
raise InvoiceValidationError(f"duplicate invoice_id: {invoice.invoice_id}")
invoices[invoice.invoice_id] = invoice
return {"invoices": invoices}
return data
@model_validator(mode="after")
@pydantic_validation_boundary
def _validate_mapping_keys(self) -> Self:
for key, invoice in self.invoices.items():
if key != invoice.invoice_id:
raise InvoiceValidationError(f"catalogue key {key!r} does not match invoice_id {invoice.invoice_id!r}")
return self
@field_validator("invoices")
@classmethod
def _freeze_invoices(cls, value: Mapping[str, Invoice]) -> Mapping[str, Invoice]:
return MappingProxyType(dict(value))
@field_serializer("invoices")
def _serialize_invoices(self, value: Mapping[str, Invoice]) -> dict[str, Invoice]:
return dict(value)
@override
def __iter__(self) -> Iterator[Invoice]: # pyright: ignore[reportIncompatibleMethodOverride] # ty: ignore[invalid-method-override] # pyrefly: ignore[bad-override] # reason: intentional Pydantic catalogue iteration adapter; the established public API yields Invoice records, not BaseModel field-value tuples
"""Iterate over catalogue invoices."""
return iter(self.invoices.values())
def __len__(self) -> int:
"""Return the number of invoices in the catalogue."""
return len(self.invoices)
def __contains__(self, invoice_id: object) -> bool:
"""Return whether the catalogue contains ``invoice_id``."""
if isinstance(invoice_id, Invoice):
return invoice_id.invoice_id in self.invoices
if isinstance(invoice_id, str):
return invoice_id in self.invoices
return False
[docs]
def get(self, invoice_id: str) -> Invoice | None:
"""Fetch one invoice by ID if present.
Args:
invoice_id: Stable invoice identifier.
Returns:
The matching :class:`Invoice`, or ``None`` when absent.
"""
return self.invoices.get(invoice_id)
[docs]
def values(self) -> Iterator[Invoice]:
"""Iterate over catalogue :class:`Invoice` records."""
return iter(self.invoices.values())