teff.graph.conditions

teff.graph.conditions

Edge condition parsing and evaluation.

Conditions are lightweight string expressions attached to edges. They are evaluated against the current workflow state to decide which node runs next. Splitting them out from the :class:~teff.graph.Graph class keeps the execution engine free of expression-language details.

Functions:

Name	Description
evaluate	Whether condition matches state.
find_error_edge	Return the __error__ edge leaving node_id, if any.
matched_condition	Return the condition of the first edge matching state and target_id.
resolve_edge	Return the target of the first edge matching state, or None.

evaluate

evaluate(condition, state)

Whether condition matches state.

Supports equality/inequality on string keys, comma-separated disjunctions (key=a,b), numeric comparisons (key>=N / key<=N / key>N / key<N), and callable predicates fn(state) -> bool (evaluated verbatim).

Source code in teff/graph/conditions.py

def evaluate(condition: str | Callable[[dict], bool], state: dict) -> bool:
    """Whether *condition* matches *state*.

    Supports equality/inequality on string keys, comma-separated
    disjunctions (``key=a,b``), numeric comparisons
    (``key>=N`` / ``key<=N`` / ``key>N`` / ``key<N``), and callable
    predicates ``fn(state) -> bool`` (evaluated verbatim).
    """
    if callable(condition):
        return bool(condition(state))
    parts = _split_condition(condition)
    if parts is None:
        return False
    op, key, raw = parts
    state_val = state.get(key)

    if op in (">=", "<=", ">", "<"):
        if state_val is None:
            return False
        try:
            left = float(state_val)
            right = float(raw)
        except (TypeError, ValueError):
            return False
        return {">": _gt, "<": _lt, ">=": _gte, "<=": _lte}[op](left, right)

    if op == "!=":
        if raw == "":
            return state_val is not None and state_val != ""
        if state_val is None:
            return True
        state_str = _norm(str(state_val))
        if "," in raw:
            values = [_norm(v) for v in raw.split(",")]
            return state_str not in values
        return state_str != _norm(raw)

    # Equality ("=").
    if raw == "":
        return state_val is None or state_val == ""
    if state_val is None:
        return False
    state_str = _norm(str(state_val))
    if "," in raw:
        values = [_norm(v) for v in raw.split(",")]
        return state_str in values
    return state_str == _norm(raw)

find_error_edge

find_error_edge(edges, node_id)

Return the __error__ edge leaving node_id, if any.

Source code in teff/graph/conditions.py

def find_error_edge(edges: list[Edge], node_id: str) -> Edge | None:
    """Return the ``__error__`` edge leaving *node_id*, if any."""
    for e in edges:
        if e.source_id == node_id and e.condition == _ERROR_CONDITION:
            return e
    return None

matched_condition

matched_condition(edges, state, target_id)

Return the condition of the first edge matching state and target_id.

Source code in teff/graph/conditions.py

def matched_condition(
    edges: list[Edge], state: dict, target_id: str
) -> "str | Callable[[dict], bool] | None":
    """Return the condition of the first edge matching *state* and *target_id*."""
    for edge in edges:
        if edge.target_id != target_id:
            continue
        if edge.condition is None or evaluate(edge.condition, state):
            return edge.condition
    return None

resolve_edge

resolve_edge(edges, state)

Return the target of the first edge matching state, or None.

Source code in teff/graph/conditions.py

def resolve_edge(edges: list[Edge], state: dict) -> str | None:
    """Return the target of the first edge matching *state*, or ``None``."""
    for edge in edges:
        if edge.condition is None:
            return edge.target_id
        if evaluate(edge.condition, state):
            return edge.target_id
    return None