Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
35 changes: 28 additions & 7 deletions src/determa/state/cel.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,28 +12,47 @@
from __future__ import annotations

from functools import lru_cache
from typing import Any
from typing import TYPE_CHECKING, Any, cast

import celpy
import celpy.celtypes as celtypes
if TYPE_CHECKING:
import celpy


class CelError(Exception):
"""A CEL expression failed to compile or evaluate (e.g. division by zero)."""


_env = celpy.Environment()
# celpy (with its lark + pendulum transitive deps) costs ~0.1s to import, but is only
# needed to *evaluate* an expression — not to load, inspect, snapshot, or step a machine.
# Import it lazily so the CLI and library stay fast on guard-free paths (SPEC §6).
_celpy: Any = None
_celtypes: Any = None
_env: Any = None


def _load() -> tuple[Any, Any, Any]:
global _celpy, _celtypes, _env
if _celpy is None:
import celpy as celpy_mod
import celpy.celtypes as celtypes_mod

_celpy = celpy_mod
_celtypes = celtypes_mod
_env = celpy_mod.Environment()
return _celpy, _celtypes, _env


@lru_cache(maxsize=2048)
def _program(expr: str) -> celpy.Runner:
celpy_mod, _, env = _load()
try:
return _env.program(_env.compile(expr))
except celpy.CELEvalError as exc: # type: ignore[attr-defined]
return cast("celpy.Runner", env.program(env.compile(expr)))
except celpy_mod.CELEvalError as exc:
raise CelError(f"compile error: {expr!r}: {exc}") from exc


def _to_cel(value: Any) -> Any:
_, celtypes, _ = _load()
if isinstance(value, dict):
return celtypes.MapType({k: _to_cel(v) for k, v in value.items()})
if isinstance(value, list):
Expand All @@ -48,6 +67,7 @@ def _from_cel(value: Any) -> Any:
so every CEL result is coerced to its native equivalent here — the single choke point
for esv assignments, published payloads, and spawn args.
"""
_, celtypes, _ = _load()
if isinstance(value, celtypes.BoolType): # subclasses int — check before IntType
return bool(value)
if isinstance(value, (celtypes.IntType, celtypes.UintType)):
Expand All @@ -67,7 +87,8 @@ def _from_cel(value: Any) -> Any:

def evaluate(expr: str, bindings: dict[str, Any]) -> Any:
"""Evaluate a CEL expression, returning a canonical native/JSON value (§5.1)."""
celpy_mod, _, _ = _load()
try:
return _from_cel(_program(expr).evaluate(_to_cel(bindings)))
except celpy.CELEvalError as exc: # type: ignore[attr-defined]
except celpy_mod.CELEvalError as exc:
raise CelError(str(exc)) from exc
4 changes: 2 additions & 2 deletions src/determa/state/validator.py
Original file line number Diff line number Diff line change
Expand Up @@ -26,8 +26,6 @@
from pathlib import Path
from typing import Any, cast

import jsonschema

from .errors import ErrorRecord, ValidationError

RESERVED_NAMES = frozenset({"top", "id", "parent", "event"})
Expand Down Expand Up @@ -67,6 +65,8 @@ def collect_errors(doc: dict[str, Any]) -> list[ErrorRecord]:


def _structural_errors(doc: dict[str, Any]) -> list[ErrorRecord]:
import jsonschema # deferred: ~40ms to import, only needed when validating a machine

validator = jsonschema.Draft202012Validator(schema())
out: list[ErrorRecord] = []
for err in sorted(validator.iter_errors(doc), key=lambda e: list(e.absolute_path)):
Expand Down
Loading