"""Retenciones-aggregation registry binding helpers (RET-1).
The per-family module for the ``retenciones_aggregation`` binding source — the
calc-mesh source that materialises retenciones-family scalars from the dedicated
per-perceptor retención store.
Extracted as its own family module per ``aeat-architecture-boundaries``;
the cross-family validator dispatch in :mod:`.bindings` registers this family's
``validate(binding) -> list[str]`` entry.
The materialisation depends on a :class:`_RetencionesAggregationProtocol` rather
than the application-layer ``RetencionesAggregation``, so the domain layer stays
independent of the application aggregation service.
The resolver walks the declared :class:`ModeloRevision` bindings and emits only
canonical binding values for this source family.
"""
from __future__ import annotations
from decimal import Decimal
from enum import StrEnum
from typing import TYPE_CHECKING, Annotated, Literal, Protocol
from pydantic import BaseModel, BeforeValidator
from ....core.aggregation import BindingSourceKind, RetencionScheme
from ....core.casilla_id import CasillaId
from ....core.models import STRICT_FROZEN_CONFIG
from .binding_selector_utils import provider_member, selector_against_model
from .errors import RegistryValidationError
from .governed_fact_scope import GovernedFactSource, governed_facts_in_scope
from .ids import BindingId
from .schema_base import coerce_enum_member, coerce_enum_tuple
if TYPE_CHECKING:
from ....core.period import Period
from .schema import BindingDefinition, ModeloRevision
[docs]
class RetencionesAggregationFact(StrEnum):
"""A figure a retenciones aggregation binding totals across matched rows."""
PERCEPTOR_COUNT_DISTINCT = "perceptor_count_distinct"
TAXABLE_BASE_SUM = "taxable_base_sum"
RETENCION_AMOUNT_SUM = "retencion_amount_sum"
RetencionesAggregationFactField = Annotated[
RetencionesAggregationFact, BeforeValidator(coerce_enum_member(RetencionesAggregationFact))
]
"""Registry token hydrated into a RetencionesAggregationFact member."""
class _RetencionesAggregationProtocol(Protocol):
"""The scalar totals the retenciones-aggregation source materialises."""
@property
def modelo(self) -> str: ...
@property
def period(self) -> Period: ...
@property
def rollups(self) -> tuple[_RetencionesRollupProtocol, ...]: ...
@property
def total_perceptors(self) -> int: ...
@property
def total_taxable_base(self) -> Decimal: ...
@property
def total_retencion(self) -> Decimal: ...
class _RetencionesRollupProtocol(Protocol):
"""One per-perceptor/scheme rollup row exposed by the retenciones source."""
@property
def perceptor_nif(self) -> str: ...
@property
def scheme(self) -> RetencionScheme: ...
@property
def total_taxable_base(self) -> Decimal: ...
@property
def total_retencion(self) -> Decimal: ...
[docs]
class RetencionesAggregationProvider(BaseModel):
"""Validated form of a ``retenciones_aggregation`` binding selector.
Carries the target casilla id and the scalar fact this source serves. Annual
summary modelos still keep their monetary totals on relation-prefill
bindings; Modelo 115 uses the same per-perceptor store directly for its
quarterly perceptor count and taxable base.
"""
model_config = STRICT_FROZEN_CONFIG
kind: Literal[BindingSourceKind.RETENCIONES_AGGREGATION] = BindingSourceKind.RETENCIONES_AGGREGATION
target_casilla_id: CasillaId
schemes: Annotated[tuple[RetencionScheme, ...], BeforeValidator(coerce_enum_tuple(RetencionScheme))] = ()
fact: RetencionesAggregationFactField = RetencionesAggregationFact.PERCEPTOR_COUNT_DISTINCT
[docs]
def validate_retenciones_aggregation_binding(binding: BindingDefinition) -> list[str]:
"""Accumulating registry-build validator for a ``retenciones_aggregation`` binding.
Validates the selector shape against :class:`RetencionesAggregationProvider`
(the single build-time contract per ``aeat-registry-bindings``),
preserving the underlying pydantic field message in the diagnostic.
"""
return selector_against_model(binding, RetencionesAggregationProvider)
[docs]
def resolve_retenciones_aggregation_binding_values(
revision: ModeloRevision,
aggregation: _RetencionesAggregationProtocol,
*,
authority: GovernedFactSource | None = None,
) -> dict[BindingId, Decimal]:
"""Materialise every ``retenciones_aggregation`` binding on a :class:`ModeloRevision`.
Returns a binding value for every binding whose ``source`` is
:attr:`BindingSourceKind.RETENCIONES_AGGREGATION`, keyed by the selector's
declared fact.
"""
resolved: dict[BindingId, Decimal] = {}
for binding in revision.bindings:
if binding.source != BindingSourceKind.RETENCIONES_AGGREGATION:
continue
selector = provider_member(binding, RetencionesAggregationProvider)
resolved[binding.id] = _retenciones_selector_value(selector, aggregation, authority=authority)
return resolved
def _registry_schemes_for_modelo(
aggregation: _RetencionesAggregationProtocol,
*,
authority: GovernedFactSource | None = None,
) -> frozenset[RetencionScheme]:
"""Resolve the selected modelo's allowed scheme tokens from fact authority."""
from .facts.resolution import MappingFactQuery, ResolvedMappingFact
from .schema_base import DateAxis
selected_authority = authority or governed_facts_in_scope()
if selected_authority is None:
raise RegistryValidationError("retenciones scheme lookup requires an explicit authority operation or scope")
resolved = selected_authority.resolve_governed_fact(
MappingFactQuery(
fact_id="m111-m115-m123-withholding-scheme-catalogue",
date_axis=DateAxis.FILING_PERIOD,
effective_date=aggregation.period.end_date,
),
)
if not isinstance(resolved, ResolvedMappingFact):
raise RegistryValidationError("withholding scheme catalogue did not resolve as a mapping fact")
target_key = f"modelo.{aggregation.modelo}.schemes"
declarations = [entry.value for entry in resolved.payload.entries if entry.key == target_key]
if len(declarations) != 1 or not isinstance(declarations[0], str):
raise RegistryValidationError(
f"withholding scheme catalogue must declare exactly one {target_key!r} entry",
)
tokens = tuple(token.strip() for token in declarations[0].split(",") if token.strip())
if not tokens:
raise RegistryValidationError(f"withholding scheme catalogue entry {target_key!r} is empty")
try:
schemes = frozenset(RetencionScheme(token) for token in tokens)
except ValueError as exc:
raise RegistryValidationError(f"withholding scheme catalogue entry {target_key!r} is malformed") from exc
if len(schemes) != len(tokens):
raise RegistryValidationError(f"withholding scheme catalogue entry {target_key!r} contains duplicates")
return schemes
def _retenciones_selector_value(
selector: RetencionesAggregationProvider,
aggregation: _RetencionesAggregationProtocol,
*,
authority: GovernedFactSource | None = None,
) -> Decimal:
if not selector.schemes:
values = {
"perceptor_count_distinct": Decimal(aggregation.total_perceptors),
"taxable_base_sum": aggregation.total_taxable_base,
"retencion_amount_sum": aggregation.total_retencion,
}
return values[selector.fact]
declared_schemes = _registry_schemes_for_modelo(aggregation, authority=authority)
unknown_schemes = frozenset(selector.schemes).difference(declared_schemes)
if unknown_schemes:
rendered = ", ".join(sorted(scheme.value for scheme in unknown_schemes))
raise RegistryValidationError(
f"retenciones binding declares scheme token(s) outside the selected registry catalogue: {rendered}",
)
selected = tuple(row for row in aggregation.rollups if row.scheme in selector.schemes)
if selector.fact == "perceptor_count_distinct":
return Decimal(len({row.perceptor_nif for row in selected}))
if selector.fact == "taxable_base_sum":
return sum((row.total_taxable_base for row in selected), Decimal("0"))
return sum((row.total_retencion for row in selected), Decimal("0"))
__all__ = [
"RetencionesAggregationProvider",
"resolve_retenciones_aggregation_binding_values",
"validate_retenciones_aggregation_binding",
]
RetencionesAggregationProvider = RetencionesAggregationProvider