"""Strict pydantic v2 schema for the :mod:`cadrumo.domain.iva` subpackage.
Every record the subpackage exposes — enumerations, per-rate values,
citations, regulations, catalogues, verification reports — is defined here.
The schema is frozen and strict wherever the loader idiom permits it,
matching the current registry-backed legal grounding conventions.
The IVA category, rate-kind, cash-treatment, exemption, and member-state
vocabularies are registry-projected opaque tokens. Every prose field
is stored inline and Spanish-authoritative: a citation's ``quoted_text`` is
verbatim BOE text, which is evidence rather than a label, and so has no
locale-resolved form.
"""
from __future__ import annotations
from collections.abc import Iterator
from datetime import date
from decimal import Decimal
from enum import StrEnum
from typing import TYPE_CHECKING, Annotated, Protocol, override
from pydantic import (
BaseModel,
Field,
StringConstraints,
field_serializer,
field_validator,
model_validator,
)
from ...core.citation_grounding import CitationGrounding
from ...core.errors.hierarchy import pydantic_validation_boundary
from ...core.models import STRICT_FROZEN_CONFIG
from ...core.parsing.dates import parse_iso8601_date
from ...core.validity_window import ValidityWindow
from .errors import IvaValidationError
if TYPE_CHECKING:
from ..calculations.registry.governed_fact_scope import GovernedFactSource
class _ModuloOrdenAnualLike(Protocol):
order: int
coefficient: Decimal
identity: str
[docs]
def validate_orden_module_identities(modulos: tuple[_ModuloOrdenAnualLike, ...]) -> None:
"""Refuse an Orden activity whose modules are missing, misordered or non-positive."""
if not 1 <= len(modulos) <= 7:
raise IvaValidationError("an Orden activity must declare one through seven modules")
if tuple(module.order for module in modulos) != tuple(range(1, len(modulos) + 1)):
raise IvaValidationError("an Orden activity module identities must be complete and ordered")
if any(module.coefficient <= 0 for module in modulos):
raise IvaValidationError("an Orden activity module must have a positive coefficient")
identities = tuple(module.identity for module in modulos)
if len(set(identities)) != len(identities):
raise IvaValidationError("an Orden activity contains duplicate module identities")
[docs]
class IvaCategory(str):
"""Opaque IVA-category token projected from the facts authority."""
__slots__ = ()
@classmethod
def __get_pydantic_core_schema__(cls, _source_type: object, _handler: object) -> object:
"""Expose the opaque token as a non-empty string to Pydantic."""
from pydantic_core import core_schema
return core_schema.no_info_after_validator_function(
pydantic_validation_boundary(cls), core_schema.str_schema(min_length=1)
)
@property
def value(self) -> str:
"""Return the persisted registry token for serialization."""
return str(self)
[docs]
class IvaCashAccountingTreatment(str):
"""Opaque registry-projected IVA cash-accounting treatment token.
Treatment membership and legal semantics are selected from fact 0098;
this type retains only the stable token shape.
"""
__slots__ = ()
@classmethod
def __get_pydantic_core_schema__(cls, _source_type: object, _handler: object) -> object:
"""Expose the opaque token as a non-empty string to Pydantic."""
from pydantic_core import core_schema
return core_schema.no_info_after_validator_function(
pydantic_validation_boundary(cls), core_schema.str_schema(min_length=1)
)
@property
def value(self) -> str:
"""Return the opaque token for string-oriented serialization."""
return str(self)
[docs]
class IvaLedgerObservationRole(StrEnum):
"""The reporting role of an IVA ledger observation.
This is deliberately independent from
:class:`IvaCashAccountingTreatment`. A cash-accounting operation creates
an informational art. 75 projection at its operation date and monetary
settlement projections at each payment (or the statutory fallback) date.
Both carry the same treatment affiliation, while their role tells registry
bindings which projection they are permitted to consume.
Ordinary IVA observations are monetary projections too, so they use
:attr:`SETTLEMENT`. That keeps normal IVA bindings on one role while
reserving :attr:`OPERATION_INFORMATIONAL` for Modelo 303 boxes 62, 63, 74,
and 75.
"""
OPERATION_INFORMATIONAL = "operation_informational"
SETTLEMENT = "settlement"
[docs]
class IvaCashAccountingPaymentEvidence(BaseModel):
"""Collection/payment evidence for an operation affected by criterio de caja.
The amounts are the IVA substrate settled by the collection/payment event,
not a gross bank amount that downstream code must reinterpret. Partial
evidence therefore remains explicit and auditable.
"""
model_config = STRICT_FROZEN_CONFIG
payment_date: date
taxable_base: Decimal = Field(..., ge=Decimal("0"))
iva_amount: Decimal = Field(..., ge=Decimal("0"))
recargo_amount: Decimal = Field(default=Decimal("0"), ge=Decimal("0"))
@field_validator("payment_date", mode="before")
@classmethod
@pydantic_validation_boundary
def _parse_payment_date(cls, value: object) -> object:
if isinstance(value, str):
return parse_iso8601_date(value)
return value
@field_validator("taxable_base", "iva_amount", "recargo_amount", mode="before")
@classmethod
@pydantic_validation_boundary
def _coerce_decimal_field(cls, value: object) -> object:
"""Accept a JSON-decoded ``Decimal`` string alongside a real ``Decimal``.
Without this, a ``Transaction`` carrying a populated
``cash_accounting_payment_evidence`` tuple cannot round-trip through
``Envelope[Transaction].model_validate_json`` at all: pydantic-core
strict mode rejects the JSON-decoded string for these fields even
though ``payment_date`` was already coerced above.
"""
if isinstance(value, str):
return Decimal(value)
return value
@model_validator(mode="after")
@pydantic_validation_boundary
def _require_settlement_amount(self) -> IvaCashAccountingPaymentEvidence:
if (
self.taxable_base == Decimal("0")
and self.iva_amount == Decimal("0")
and self.recargo_amount == Decimal("0")
):
raise IvaValidationError(
"cash-accounting payment evidence must carry a non-zero base, IVA, or recargo amount",
)
return self
@field_serializer("taxable_base", "iva_amount", "recargo_amount", when_used="json")
def _serialize_decimal(self, value: Decimal) -> str:
return str(value)
class _OpaqueRegistryToken(str):
"""Opaque registry-projected token exposed to Pydantic as non-empty text."""
__slots__ = ()
@classmethod
def __get_pydantic_core_schema__(cls, _source_type: object, _handler: object) -> object:
"""Expose the opaque token as a non-empty string to Pydantic."""
from pydantic_core import core_schema
return core_schema.no_info_after_validator_function(
pydantic_validation_boundary(cls), core_schema.str_schema(min_length=1)
)
@property
def value(self) -> str:
"""Return the opaque token for string-oriented serialization."""
return str(self)
[docs]
class IvaExemptionArticle(_OpaqueRegistryToken):
"""Opaque registry-projected IVA exemption-article token."""
__slots__ = ()
[docs]
class IvaArt69DosService(_OpaqueRegistryToken):
"""Opaque registry-projected Art. 69.Dos service token."""
__slots__ = ()
[docs]
class EUMemberState(str):
"""Opaque EU IVA member-state token projected from fact 0131."""
__slots__ = ()
def __new__(cls, value: str, *, _registry_validated: bool = False) -> EUMemberState:
"""Reject direct construction outside the facts-registry projection."""
if not _registry_validated:
raise TypeError("EUMemberState tokens must be projected from the facts registry")
if not isinstance(value, str) or not value:
raise ValueError("EU member-state token must be a non-empty string")
return str.__new__(cls, value)
[docs]
@classmethod
def from_registry(cls, value: str) -> EUMemberState:
"""Construct a token only at the typed registry projection boundary."""
return cls(value, _registry_validated=True)
@classmethod
def __get_pydantic_core_schema__(cls, _source_type: object, _handler: object) -> object:
"""Validate Pydantic values against the selected member-state fact."""
from pydantic_core import core_schema
return core_schema.no_info_after_validator_function(
pydantic_validation_boundary(cls._project_pydantic),
core_schema.str_schema(min_length=2, max_length=2),
)
@classmethod
def _project_pydantic(cls, value: object) -> EUMemberState:
return require_eu_member_state(value)
@property
def value(self) -> str:
"""Return the persisted registry token for serialization."""
return str(self)
[docs]
def require_eu_member_state(
value: object,
*,
effective_date: date | None = None,
authority: GovernedFactSource | None = None,
) -> EUMemberState:
"""Project one EU member-state token through fact 0131."""
from ..calculations.registry.eu_member_state_catalogue import (
require_eu_member_state as _require_eu_member_state,
)
return _require_eu_member_state(
value,
effective_date=effective_date,
authority=authority,
)
[docs]
def require_registry_declared_eu_member_state(value: object, *, effective_date: date) -> EUMemberState:
"""Project a member-state token from the facts currently under validation."""
from ..calculations.registry.eu_member_state_catalogue import (
require_registry_declared_eu_member_state as _require_registry_declared_eu_member_state,
)
return _require_registry_declared_eu_member_state(value, effective_date=effective_date)
[docs]
def spanish_eu_member_state(
*,
effective_date: date | None = None,
authority: GovernedFactSource | None = None,
) -> EUMemberState:
"""Return the registry-declared Spain token for domestic IVA projections."""
return require_eu_member_state("ES", effective_date=effective_date, authority=authority)
[docs]
class IvaRateKind(str):
"""Opaque registry-projected IVA rate-tier token."""
__slots__ = ()
@classmethod
def __get_pydantic_core_schema__(cls, _source_type: object, _handler: object) -> object:
"""Expose the opaque token as a non-empty string to Pydantic."""
from pydantic_core import core_schema
return core_schema.no_info_after_validator_function(
pydantic_validation_boundary(cls), core_schema.str_schema(min_length=1)
)
@property
def value(self) -> str:
"""Return the persisted registry token for serialization."""
return str(self)
[docs]
def default_iva_cash_accounting_treatment(
on_date: date | None = None,
*,
authority: GovernedFactSource | None = None,
) -> IvaCashAccountingTreatment:
"""Return the registry-declared ordinary cash-accounting treatment."""
from ..calculations.registry.iva_schema_vocabulary import default_iva_cash_accounting_treatment as resolve_default
return resolve_default(effective_date=on_date, authority=authority)
[docs]
def is_iva_cash_accounting_none(
treatment: IvaCashAccountingTreatment,
on_date: date | None = None,
*,
authority: GovernedFactSource | None = None,
) -> bool:
"""Return whether a treatment is the registry-declared ordinary token."""
return treatment == default_iva_cash_accounting_treatment(on_date, authority=authority)
[docs]
def is_iva_cash_accounting_supplier_regime(
treatment: IvaCashAccountingTreatment,
on_date: date | None = None,
*,
authority: GovernedFactSource | None = None,
) -> bool:
"""Return whether a treatment is the registry-declared supplier token."""
from ..calculations.registry.iva_schema_vocabulary import resolve_iva_cash_accounting_catalogue
catalogue = resolve_iva_cash_accounting_catalogue(effective_date=on_date, authority=authority)
return treatment == catalogue.supplier_regime_token
_RegistryLegalRef = Annotated[
str,
StringConstraints(
strip_whitespace=True,
min_length=1,
max_length=160,
pattern=r"^[a-z0-9][a-z0-9._:-]*[a-z0-9]$|^[a-z0-9]$",
),
]
"""Registry legal-reference identifier; catalogue verification resolves it."""
_RegistrySourceRef = Annotated[
str,
StringConstraints(
strip_whitespace=True,
min_length=1,
max_length=160,
pattern=r"^[a-z0-9][a-z0-9._:-]*[a-z0-9]$|^[a-z0-9]$",
),
]
"""Registry source-reference identifier; catalogue verification resolves it."""
_ManualRef = Annotated[
str,
StringConstraints(strip_whitespace=True, min_length=1, max_length=256),
]
"""Free-form reference to a Manual práctico IVA rule id or section reference."""
class _IvaStrictFrozen(BaseModel):
"""Shared base config: strict validation, immutable, extras forbidden."""
model_config = STRICT_FROZEN_CONFIG
IvaStrictFrozen = _IvaStrictFrozen
class _IvaStrictMutable(BaseModel):
"""Strict validation mixin with mutable config for incrementally populated catalogues."""
model_config = STRICT_FROZEN_CONFIG
def _require_grounded_rate_refs(record: IvaRateRecord) -> None:
"""Refuse duplicate, absent, or jurisdiction-inappropriate registry grounding.
A Spanish rate is established by binding domestic law, so it must cite
``legal_refs``; a foreign member-state rate is observed rather than enacted
here, so it must cite ``source_refs`` instead.
"""
label = f"IvaRateRecord[{record.member_state.value}/{record.kind.value}]"
if len(set(record.legal_refs)) != len(record.legal_refs):
raise IvaValidationError(f"{label}: legal_refs must be unique")
if len(set(record.source_refs)) != len(record.source_refs):
raise IvaValidationError(f"{label}: source_refs must be unique")
if not record.legal_refs and not record.source_refs:
raise IvaValidationError(f"{label}: missing registry legal_refs/source_refs")
if record.member_state == spanish_eu_member_state() and not record.legal_refs:
raise IvaValidationError(f"{label}: Spanish rates require binding registry legal_refs")
if record.member_state != spanish_eu_member_state() and not record.source_refs:
raise IvaValidationError(f"{label}: foreign rates require registry source_refs")
[docs]
class IvaRateRecord(_IvaStrictFrozen):
"""A single IVA rate line item keyed by member state and rate kind.
Attributes:
member_state: Issuing member state.
kind: Rate tier (general / reduced / ...).
pct: Rate percentage as a :class:`~decimal.Decimal` in ``[0, 100]``.
effective_from: First date the rate applies.
effective_until: Last date the rate applies, or ``None`` for
open-ended.
legal_refs: Binding registry legal-reference identities backing the
numerical rate where provision-level authority is bundled.
source_refs: Registry source-reference identities backing the
numerical foreign-rate evidence.
"""
member_state: EUMemberState = Field(description="Issuing member state.")
kind: IvaRateKind = Field(description="Rate tier (general / reduced / ...).")
pct: Decimal = Field(
ge=Decimal("0"),
le=Decimal("100"),
description="Rate percentage as a decimal in [0, 100].",
)
effective_from: date = Field(description="First date the rate applies.")
effective_until: date | None = Field(
default=None,
description="Last date the rate applies, or ``None`` for open-ended.",
)
legal_refs: tuple[_RegistryLegalRef, ...] = Field(
default=(),
description="Binding registry legal-reference identities backing this numerical rate.",
)
source_refs: tuple[_RegistrySourceRef, ...] = Field(
default=(),
description="Registry source-reference identities backing this rate.",
)
supersedes_tier_default: bool = Field(
default=False,
description="A rate that applies ALONGSIDE its tier's ordinary rate, not instead of it.",
)
"""Whether this rate coexists with its tier's ordinary rate rather than replacing it.
``False`` -- the ordinary case -- means the record IS the tier's rate for its
window, so :func:`lookup_rate` can answer "what does this tier mean on this
date" with one record and the no-overlap rule guarantees that answer is
unambiguous.
``True`` marks a rate that a statute applied to PART of a tier's supplies
while the rest stayed on the ordinary rate. Spain's 2024 anti-inflation
measures are the worked case: certain foodstuffs moved to 2 % while every
other super-reducido supply stayed at 4 %, so both were simultaneously
correct and neither replaced the other (RDL 4/2024 art. 1). Such a record is
excluded from the no-overlap rule and from :func:`lookup_rate`, because
including it would make the tier's rate ambiguous for the far larger set of
supplies that never moved.
The distinction is which QUESTION each record can answer. A coexisting rate
cannot answer "what is this tier's rate" -- only the statute's goods scope
decides that, and no bundled AEAT surface carries a goods axis. It can
answer "is this declared rate a legitimate one for this tier on this date",
which is what classification needs and what
:func:`rate_kinds_for_declared_rate` serves.
"""
@model_validator(mode="after")
@pydantic_validation_boundary
def _validate_window(self) -> IvaRateRecord:
"""Ensure :attr:`effective_from` precedes :attr:`effective_until`."""
if self.effective_until is not None and self.effective_from > self.effective_until:
raise IvaValidationError(
f"IvaRateRecord[{self.member_state.value}/{self.kind.value}]: "
f"effective_from {self.effective_from} is after effective_until {self.effective_until}",
)
_require_grounded_rate_refs(self)
return self
[docs]
class IvaCitation(_IvaStrictFrozen):
"""A legal or regulatory citation backing a :class:`IvaRegulation`.
:attr:`quoted_text` holds the authoritative Spanish INLINE, not a
translation key. Verifying a quotation against the bundled corpus requires
the literal text at the citation site; indirecting it means the record no
longer carries its own evidence, whatever the key resolves to.
Attributes:
legal_reference: Article-qualified registry legal-reference id.
quoted_text: Verbatim Spanish from the bundled corpus. Empty only when
:attr:`grounding` is ``UNRESOLVED``.
grounding: Whether the text was verified or the citation refused.
unresolved_reason: Why the citation could not be grounded. Required
when, and only when, :attr:`grounding` is ``UNRESOLVED``.
valid_from: First date this citation is asserted to support the rule.
valid_to: Last date, inclusive.
Both window bounds are required and closed. The span is a grounding claim,
not bookkeeping: it says the cited provision supports this rule across those
years, and a gate holds it inside the provision's own effective span in the
registry legal catalogue. A defaulted or open end would assert grounding
nobody typed and nothing could check.
"""
legal_reference: _RegistryLegalRef = Field(
description="Article-qualified id resolved through the registry legal catalogue.",
)
quoted_text: str = Field(
default="",
description="Verbatim Spanish from the bundled corpus; empty only when grounding is unresolved.",
)
grounding: CitationGrounding = Field(
default=CitationGrounding.VERIFIED,
description="Whether the quotation was verified against the corpus, or examined and refused.",
)
unresolved_reason: str = Field(
default="",
description="Why the citation could not be grounded; required when grounding is unresolved.",
)
valid_from: date = Field(
description="First date this citation is asserted to support the rule.",
)
valid_to: date = Field(
description="Last date, inclusive, this citation is asserted to support the rule.",
)
@property
def window(self) -> ValidityWindow:
"""Return the closed span this citation is asserted over.
Returns:
The :class:`~core.validity_window.ValidityWindow` the declared
bounds describe.
"""
return ValidityWindow(valid_from=self.valid_from, valid_to=self.valid_to)
@model_validator(mode="after")
@pydantic_validation_boundary
def _validate(self) -> IvaCitation:
"""Hold each grounding state to the evidence it claims.
Unlike the validator this replaced, both branches can fail. The old
one asserted a translatable was non-empty AFTER the loader had already
resolved it through a fallback that never yields an empty string, so
it inspected ``"Quoted text"``, found it non-empty, and passed for
every citation in the catalogue.
"""
where = f"IvaCitation[{self.legal_reference}]"
# Constructing the window is the validation: an inverted span covers no
# year at all, and must refuse where it was written rather than make the
# citation silently vanish from every filing year downstream.
_ = self.window
if self.grounding is CitationGrounding.VERIFIED:
if not self.quoted_text.strip():
raise IvaValidationError(f"{where}: a verified citation must carry its verbatim quotation")
if self.unresolved_reason.strip():
raise IvaValidationError(f"{where}: a verified citation must not carry an unresolved reason")
else:
if not self.unresolved_reason.strip():
raise IvaValidationError(
f"{where}: an unresolved citation must record WHY it could not be grounded, "
"so that it reads as examined and refused rather than merely unchecked",
)
if self.quoted_text.strip():
# Text parked here would never be read against the corpus
# while the record says it could not be.
raise IvaValidationError(
f"{where}: an unresolved citation must not carry a quotation; "
"text that survived the corpus read belongs under verified grounding",
)
return self
[docs]
class IvaRegulation(_IvaStrictFrozen):
"""A single codified IVA rule for a :class:`IvaCategory`.
Every regulation carries at least one :class:`IvaCitation` so
downstream tools can surface the legal backing of any classification.
Attributes:
category: The IVA situation codified by this rule.
requires_reverse_charge: Whether the rule triggers
*inversión del sujeto pasivo*.
requires_supplier_iva_id: Whether a supplier NIF-IVA is mandatory.
manual_references: Optional Manual práctico IVA rule ids or section
references.
citations: At least one :class:`IvaCitation` is required, unless
:attr:`legal_basis_exempt` is set.
notes: Free-form reviewer notes.
legal_basis_exempt: True only for a category that codifies no tax
treatment at all (an application-level classifier sentinel), so
citing law for it would manufacture the appearance of a legal
basis a construct without one by design. Citing an unrelated or
over-broad article is not this: that is a wrong citation on a
category that DOES need grounding, and must be fixed, not
exempted.
"""
category: IvaCategory = Field(description="The IVA situation codified by this rule.")
requires_reverse_charge: bool = Field(
description="Whether the rule triggers inversión del sujeto pasivo.",
)
requires_supplier_iva_id: bool = Field(
description="Whether a supplier NIF-IVA is mandatory for this rule.",
)
manual_references: tuple[_ManualRef, ...] = Field(
description="Optional Manual práctico IVA rule ids or section refs.",
)
citations: tuple[IvaCitation, ...] = Field(
description="At least one IvaCitation is required, unless legal_basis_exempt is set.",
)
notes: str = Field(
default="",
description="Free-form reviewer notes.",
)
legal_basis_exempt: bool = Field(
default=False,
description="True only for a classifier sentinel that codifies no tax treatment.",
)
@model_validator(mode="after")
@pydantic_validation_boundary
def _validate(self) -> IvaRegulation:
"""Enforce the at-least-one-citation invariant and its sole carve-out."""
if self.legal_basis_exempt:
if self.citations:
raise IvaValidationError(
f"IvaRegulation[{self.category.value}]: legal_basis_exempt must carry no citations",
)
if not self.notes.strip():
raise IvaValidationError(
f"IvaRegulation[{self.category.value}]: legal_basis_exempt requires notes explaining why",
)
elif not self.citations:
raise IvaValidationError(f"IvaRegulation[{self.category.value}]: at least one IvaCitation is required")
return self
[docs]
class IvaCatalogue(_IvaStrictMutable):
"""Aggregate view over a collection of :class:`IvaRegulation` records.
The aggregate is mutable to keep the loader idiom simple — the loader
populates the mapping incrementally. Individual :class:`IvaRegulation`
records remain frozen.
Attributes:
regulations: Regulations keyed by their
:class:`IvaCategory`.
"""
regulations: dict[IvaCategory, IvaRegulation] = Field(
default_factory=dict,
description="Regulations keyed by their IvaCategory.",
)
@model_validator(mode="after")
@pydantic_validation_boundary
def _check_key_alignment(self) -> IvaCatalogue:
"""Ensure every mapping key matches its record's :attr:`IvaRegulation.category`."""
for key, regulation in self.regulations.items():
if key != regulation.category:
raise IvaValidationError(
f"IvaCatalogue: key {key!r} does not match regulation.category {regulation.category!r}",
)
return self
@override
def __iter__(self) -> Iterator[IvaRegulation]: # pyright: ignore[reportIncompatibleMethodOverride] # ty: ignore[invalid-method-override] # pyrefly: ignore[bad-override] # reason: intentional Pydantic catalogue iteration adapter; the established public API yields IvaRegulation records, not BaseModel field-value tuples
"""Iterate over every loaded :class:`IvaRegulation`."""
return iter(self.regulations.values())
def __len__(self) -> int:
"""Return the number of loaded :class:`IvaRegulation` records."""
return len(self.regulations)
def __contains__(self, key: object) -> bool:
"""Return ``True`` when ``key`` names a loaded :class:`IvaCategory`."""
return key in self.regulations
[docs]
def get(self, category: IvaCategory) -> IvaRegulation | None:
"""Return the :class:`IvaRegulation` for ``category`` or ``None`` if absent."""
return self.regulations.get(category)