Source code for aeat.domain.calculations.registry._validate_formulas

"""Formula expression and dependency graph validation helpers.

Validates formula expressions and the DAG formed by formula targets
declared on a :class:`~aeat.domain.calculations.registry.ModeloRevision`,
checking casilla, binding, parameter, and relation reference closure and
detecting cycles.

See Also:
    :func:`aeat.domain.calculations.registry._runtime_graph.expression_casilla_refs`
        Formula-expression walker used to derive target dependencies.
    :func:`aeat.domain.calculations.registry._runtime_graph.formula_evaluation_order`
        Runtime topological order builder that assumes this validator has
        rejected cycles.
"""

from __future__ import annotations

from graphlib import CycleError, TopologicalSorter

from ._ids import BindingId, CasillaId, RelationId
from ._runtime_graph import expression_casilla_refs
from ._schema import FormulaExpression, ModeloRevision


[docs] def validate_formula_dag(scope: str, revision: ModeloRevision) -> list[str]: """Return dependency-cycle failures for a revision's computed formulas. The :class:`~aeat.domain.calculations.registry.ModeloRevision` supplies formula targets and expressions. Only dependencies that point at another computed target participate in the DAG; registry membership and reference existence are handled by :func:`validate_formula_expression`. """ formula_targets = {formula.target_casilla_id for formula in revision.formulas} sorter: TopologicalSorter[str] = TopologicalSorter() for formula in revision.formulas: dependencies = [ casilla for casilla in expression_casilla_refs(formula.expression) if casilla in formula_targets ] sorter.add(formula.target_casilla_id, *dependencies) try: tuple(sorter.static_order()) except CycleError as exc: return [f"{scope}: formula graph cycle: {exc}"] return []
[docs] def validate_formula_expression( scope: str, formula_id: str, expression: FormulaExpression, *, casillas: set[CasillaId], bindings: set[BindingId], parameters: set[str], relations: set[RelationId], ) -> list[str]: """Return reference-closure failures for one formula expression tree. The :class:`~aeat.domain.calculations.registry.FormulaExpression` may refer to :class:`~aeat.domain.calculations.registry.CasillaId`, :class:`~aeat.domain.calculations.registry.BindingId`, parameter, and :class:`~aeat.domain.calculations.registry.RelationId` values. This recursive validator keeps every nested expression node inside the selected revision's declared id sets. """ failures: list[str] = [] if expression.casilla_id is not None and expression.casilla_id not in casillas: failures.append(f"{scope}: formula {formula_id!r} references unknown casilla {expression.casilla_id!r}") if expression.binding is not None and expression.binding not in bindings: failures.append(f"{scope}: formula {formula_id!r} references unknown binding {expression.binding!r}") if expression.parameter is not None and expression.parameter not in parameters: failures.append(f"{scope}: formula {formula_id!r} references unknown parameter {expression.parameter!r}") if expression.dispatch_table: for key, dispatched in expression.dispatch_table.items(): if dispatched not in parameters: failures.append( f"{scope}: formula {formula_id!r} dispatch_table[{key!r}] " f"references unknown parameter {dispatched!r}", ) if expression.relation is not None and expression.relation not in relations: failures.append(f"{scope}: formula {formula_id!r} references unknown relation {expression.relation!r}") for arg in expression.args: failures.extend( validate_formula_expression( scope, formula_id, arg, casillas=casillas, bindings=bindings, parameters=parameters, relations=relations, ), ) return failures