Skip to content

teff.node.command_node

teff.node.command_node

Declarative command node — route the graph from YAML state.

The :class:CommandNode is the YAML surface for :class:Command routing (goto / STOP). It lives in its own module so that :mod:teff.node.command (imported by the base :class:~teff.node.node.Node) never needs to import the node base class itself — avoiding a circular import.

Classes:

Name Description
CommandNode

Declarative command node: route the graph from YAML state.

CommandNode

Bases: Node

Declarative command node: route the graph from YAML state.

Returns a :class:~teff.node.command.Command whose goto is chosen from routes (the first route whose when condition matches state, using the same expressions as edges: conditions) and falls back to goto. update merges state keys after routing (reducers apply).

Use STOP as a target to terminate the run::

- id: route
  type: command
  config:
    routes:
      - when: score >= 0.8
        goto: approve
      - when: score < 0.3
        goto: reject
    goto: review
    update: {routed: true}
Source code in teff/node/command_node.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
class CommandNode(Node):
    """Declarative ``command`` node: route the graph from YAML state.

    Returns a :class:`~teff.node.command.Command` whose ``goto`` is chosen
    from ``routes`` (the first route whose ``when`` condition matches
    *state*, using the same expressions as ``edges:`` conditions) and falls
    back to ``goto``.  ``update`` merges state keys after routing (reducers
    apply).

    Use ``STOP`` as a target to terminate the run::

        - id: route
          type: command
          config:
            routes:
              - when: score >= 0.8
                goto: approve
              - when: score < 0.3
                goto: reject
            goto: review
            update: {routed: true}
    """

    type = "command"

    async def execute(self, ctx, state: dict) -> Command:
        from teff.graph.conditions import evaluate

        goto: str | object | None = None
        for route in self.config.get("routes", []) or []:
            when = route.get("when")
            if when and evaluate(when, state):
                goto = _resolve_target(route.get("goto"))
                break
        if goto is None:
            goto = _resolve_target(self.config.get("goto"))
        return Command(update=dict(self.config.get("update") or {}), goto=goto)