teff.hooks

teff.hooks

Declarative hook wiring for the hooks: workflow block.

Hooks are Python callables, so a YAML hooks: block cannot define them inline. Instead it references named hooks from a shared registry that plugins and scripts populate with :func:register (or the :func:hook decorator):

from teff import hooks

@hooks.hook("telemetry")
def telemetry(node_id, node, state, **kw):
    metrics.counter("graph.node", node_id=node_id)

hooks:
  on_node_start: telemetry
  on_node_error: [telemetry, on_error]

The resolved mapping is passed to graph.run(hooks=...). Hook callables may be sync or async. When several hooks share one event, they run in registration order.

Functions:

Name	Description
hook	Decorator form of :func:register.
register	Register fn under name (last registration wins).
resolve_hooks	Resolve a hooks: YAML block into a graph.run(hooks=...) dict.

hook

hook(name)

Decorator form of :func:register.

Source code in teff/hooks.py

def hook(name: str) -> Callable:
    """Decorator form of :func:`register`."""

    def deco(fn: Callable) -> Callable:
        return register(name, fn)

    return deco

register

register(name, fn)

Register fn under name (last registration wins).

Source code in teff/hooks.py

def register(name: str, fn: Callable) -> Callable:
    """Register *fn* under *name* (last registration wins)."""
    if not callable(fn):
        raise ConfigError(f"hook {name!r} must be callable")
    _HOOK_REGISTRY[name] = fn
    return fn

resolve_hooks

resolve_hooks(block)

Resolve a hooks: YAML block into a graph.run(hooks=...) dict.

Each event key (on_node_start/on_node_end/on_node_error) is a hook name or a list of names. An unset or null kind is skipped. Returns None for an empty block.

Source code in teff/hooks.py

def resolve_hooks(block: dict | None) -> dict | None:
    """Resolve a ``hooks:`` YAML block into a ``graph.run(hooks=...)`` dict.

    Each event key (``on_node_start``/``on_node_end``/``on_node_error``)
    is a hook name or a list of names.  An unset or ``null`` kind is
    skipped.  Returns ``None`` for an empty block.
    """
    if not isinstance(block, dict) or not block:
        return None
    out: dict[str, Callable] = {}
    for kind in KINDS:
        value = block.get(kind)
        if value is None:
            continue
        out[kind] = _compose(_names_from(value))
    return out or None