plaita.core.expression

结构化表达式引擎：FunctionCategory、FunctionDescriptor、ExpressionRegistry、ExpressionEvaluator，以及 60+ 内置函数的默认注册表工厂。

expression

plaita.core.expression — Structured expression engine.

Provides: - FunctionCategory enum for classifying expression functions. - FunctionDescriptor frozen dataclass with side-effect metadata. - ExpressionRegistry for querying functions by category / side-effect. - ExpressionEvaluator that delegates to plaita.io evaluate/parse_function. - get_default_expression_registry() factory with all 90+ built-in functions.

FunctionCategory

Bases: str, Enum

Classification categories for expression functions.

FunctionDescriptor dataclass

FunctionDescriptor(name: str, func: Callable, category: FunctionCategory, has_side_effects: bool = False, fresh_producer: bool = False, description: str = '')

Metadata wrapper around a registered expression function.

属性：

名称	类型	描述
name	str	Function name (used in expressions as $F.name(...)).
func	Callable	The underlying callable.
category	FunctionCategory	Classification category.
has_side_effects	bool	True if the function mutates its input. Side-effect functions (e.g. pop, set, delete, clear) are not thread-safe and must not be used in concurrent execution contexts without external synchronisation. Since plaita#69 the evaluator enforces a runtime gate on this flag: any argument of such a mutator that is a live $INPUT/$NODE/$GLOBAL/ $PARENT/$ENV path (or an undeclared producer call) is refused with a named error before evaluation; a shallow producer result stored as a value argument is detached (deep-copied).
fresh_producer	bool	True if the function always returns a new object whose elements cannot alias the caller's data (plaita#69). A declared producer is treated as DEEP by the gate; the parser's built-in classification (deep / shallow / live-reference) covers the default registry. The default is False so a forgotten declaration blocks the call instead of silently allowing a write-through.
description	str	Optional human-readable description.

ExpressionRegistry

ExpressionRegistry()

Registry for expression functions with category and side-effect metadata.

源代码位于： plaita/core/expression.py

def __init__(self) -> None:
    self._functions: dict[str, FunctionDescriptor] = {}

register

register(name: str, func: Callable, category: FunctionCategory, *, has_side_effects: bool = False, fresh_producer: bool = False, description: str = '', override: bool = False) -> None

Register an expression function.

参数：

名称	类型	描述	默认
name	str	Function name (used in expressions as $F.name(...)).	必需
func	Callable	The callable to execute.	必需
category	FunctionCategory	Classification category.	必需
has_side_effects	bool	True if the function mutates its input.	False
fresh_producer	bool	True if the function always returns a new object whose elements cannot alias the caller's data (plaita#69) — i.e. a deep producer, not merely a new container. Defaults to False: an undeclared producer used as any argument of a mutator is refused (fail-closed).	False
description	str	Human-readable description.	''
override	bool	If True, overwrite an existing registration with the same name instead of raising ValueError. Use this for intentional re-registration (e.g. backward-compatible register_function / dict-style assignment) rather than reaching into _functions from outside the class.	False

引发：

类型	描述
ValueError	If name is empty, or already registered with override=False.

源代码位于： plaita/core/expression.py

def register(
    self,
    name: str,
    func: Callable,
    category: FunctionCategory,
    *,
    has_side_effects: bool = False,
    fresh_producer: bool = False,
    description: str = "",
    override: bool = False,
) -> None:
    """Register an expression function.

    Args:
        name: Function name (used in expressions as ``$F.name(...)``).
        func: The callable to execute.
        category: Classification category.
        has_side_effects: ``True`` if the function mutates its input.
        fresh_producer: ``True`` if the function always returns a new
            object **whose elements cannot alias** the caller's data
            (plaita#69) — i.e. a *deep* producer, not merely a new
            container.  Defaults to ``False``: an undeclared producer used
            as any argument of a mutator is refused (fail-closed).
        description: Human-readable description.
        override: If ``True``, overwrite an existing registration with the
            same name instead of raising ``ValueError``.  Use this for
            intentional re-registration (e.g. backward-compatible
            ``register_function`` / dict-style assignment) rather than
            reaching into ``_functions`` from outside the class.

    Raises:
        ValueError: If *name* is empty, or already registered with
            ``override=False``.
    """
    if not name:
        raise ValueError("Function name must not be empty")
    if name in self._functions and not override:
        raise ValueError(f"Function '{name}' is already registered")
    self._functions[name] = FunctionDescriptor(
        name=name,
        func=func,
        category=category,
        has_side_effects=has_side_effects,
        fresh_producer=fresh_producer,
        description=description,
    )

get

get(name: str) -> Optional[FunctionDescriptor]

Look up a function descriptor by name.

源代码位于： plaita/core/expression.py

def get(self, name: str) -> Optional[FunctionDescriptor]:
    """Look up a function descriptor by name."""
    return self._functions.get(name)

unregister

unregister(name: str) -> None

Remove a function registration. No-op if name is not registered.

源代码位于： plaita/core/expression.py

def unregister(self, name: str) -> None:
    """Remove a function registration. No-op if *name* is not registered."""
    self._functions.pop(name, None)

get_callable

get_callable(name: str) -> Optional[Callable]

Look up just the callable by name (for backward compatibility).

源代码位于： plaita/core/expression.py

def get_callable(self, name: str) -> Optional[Callable]:
    """Look up just the callable by name (for backward compatibility)."""
    desc = self._functions.get(name)
    return desc.func if desc else None

names

names() -> list

已注册的函数名清单（供报错提示/补全）。

源代码位于： plaita/core/expression.py

def names(self) -> list:
    """已注册的函数名清单（供报错提示/补全）。"""
    return list(self._functions.keys())

by_category

by_category(category: FunctionCategory) -> List[FunctionDescriptor]

Return all functions in category.

源代码位于： plaita/core/expression.py

def by_category(self, category: FunctionCategory) -> List[FunctionDescriptor]:
    """Return all functions in *category*."""
    return [d for d in self._functions.values() if d.category == category]

side_effect_functions

side_effect_functions() -> List[FunctionDescriptor]

Return all functions marked with side effects.

源代码位于： plaita/core/expression.py

def side_effect_functions(self) -> List[FunctionDescriptor]:
    """Return all functions marked with side effects."""
    return [d for d in self._functions.values() if d.has_side_effects]

all_functions

all_functions() -> dict[str, FunctionDescriptor]

Return the full registry as a read-only snapshot.

源代码位于： plaita/core/expression.py

def all_functions(self) -> dict[str, FunctionDescriptor]:
    """Return the full registry as a read-only snapshot."""
    return dict(self._functions)

ExpressionEvaluator

ExpressionEvaluator(registry: Optional[ExpressionRegistry] = None)

Evaluates expressions against an execution context.

This class wraps the existing plaita.io.evaluate / parse_function logic and exposes the structured ExpressionRegistry. The actual parsing and evaluation is delegated to the battle-tested functions in plaita.io to guarantee identical results.

源代码位于： plaita/core/expression.py

def __init__(self, registry: Optional[ExpressionRegistry] = None) -> None:
    self._registry = registry or get_default_expression_registry()

evaluate

evaluate(value: Any, context: dict[str, Any], prefix: str = '$') -> Any

Evaluate a value, resolving any expressions.

Delegates to plaita.io.evaluate but passes this evaluator's ExpressionRegistry so that $F.func(...) calls resolve against the scoped registry. When the registry is the default one this is behavior-preserving; when a custom registry is supplied, function lookups go through it instead of the module-level global.

源代码位于： plaita/core/expression.py

def evaluate(
    self,
    value: Any,
    context: dict[str, Any],
    prefix: str = "$",
) -> Any:
    """Evaluate a value, resolving any expressions.

    Delegates to ``plaita.io.evaluate`` but passes this evaluator's
    ``ExpressionRegistry`` so that ``$F.func(...)`` calls resolve against
    the scoped registry.  When the registry is the default one this is
    behavior-preserving; when a custom registry is supplied, function
    lookups go through it instead of the module-level global.
    """
    from plaita.io import evaluate as _io_evaluate
    return _io_evaluate(value, context, prefix, registry=self._registry)

get_default_expression_registry

get_default_expression_registry() -> ExpressionRegistry

Return the module-level default registry with all built-in functions.

The registry is created lazily on first access and cached.

源代码位于： plaita/core/expression.py

def get_default_expression_registry() -> ExpressionRegistry:
    """Return the module-level default registry with all built-in functions.

    The registry is created lazily on first access and cached.
    """
    global _default_registry
    if _default_registry is None:
        _default_registry = _build_default_registry()
    return _default_registry