Skip to content

fastflowtransform.hooks.registry

fft_hook 

fft_hook(name=None, when=None)

Decorator to register a Python hook.

Usage:

from fastflowtransform.hooks.registry import fft_hook

@fft_hook(name="python_banner")               # no 'when' -> wildcard
def on_run_start(ctx: dict[str, Any]):
    ...

@fft_hook(name="python_banner", when="on_run_start")
def banner_for_run_start(ctx: dict[str, Any]):
    ...
  • name: logical hook name (matches project.yml hooks: ... name:). If omitted, defaults to the function name.
  • when: lifecycle event ("on_run_start", "on_run_end", "before_model", "after_model", etc.). If omitted, the hook is registered for the wildcard phase "*".
Source code in src/fastflowtransform/hooks/registry.py 
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
53
def fft_hook(name: str | None = None, when: str | None = None) -> Callable:
    """
    Decorator to register a Python hook.

    Usage:

        from fastflowtransform.hooks.registry import fft_hook

        @fft_hook(name="python_banner")               # no 'when' -> wildcard
        def on_run_start(ctx: dict[str, Any]):
            ...

        @fft_hook(name="python_banner", when="on_run_start")
        def banner_for_run_start(ctx: dict[str, Any]):
            ...

    - `name`: logical hook name (matches project.yml `hooks: ... name:`).
              If omitted, defaults to the function name.
    - `when`: lifecycle event ("on_run_start", "on_run_end",
              "before_model", "after_model", etc.).
              If omitted, the hook is registered for the wildcard phase "*".
    """

    def decorator(fn: Callable[[HookContext], Any]) -> Callable[[HookContext], Any]:
        hook_name = name or fn.__name__
        phase = when or "*"  # wildcard by default

        key = (phase, hook_name)
        if key in _HOOKS:
            raise ValueError(f"Hook already registered for {key!r}")

        _HOOKS[key] = fn
        return fn

    return decorator

resolve_hook 

resolve_hook(when, name)

Retrieve a previously-registered hook function.

Resolution order
  1. Exact match: (when, name)
  2. Wildcard match: ('*', name)

Raises KeyError if not found.

Source code in src/fastflowtransform/hooks/registry.py 
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
def resolve_hook(when: str, name: str) -> Callable[[HookContext], Any]:
    """
    Retrieve a previously-registered hook function.

    Resolution order:
      1. Exact match:   (when, name)
      2. Wildcard match: ('*', name)

    Raises KeyError if not found.
    """
    key = (when, name)
    if key in _HOOKS:
        return _HOOKS[key]

    wildcard_key = ("*", name)
    if wildcard_key in _HOOKS:
        return _HOOKS[wildcard_key]

    raise KeyError(f"No hook registered for when={when!r}, name={name!r}")

load_project_hooks 

load_project_hooks(project_dir)

Load all Python files under <project_dir>/hooks/**.py.

This executes the modules (without requiring them to be proper Python packages), so any @fft_hook(...) calls will populate the registry.

This is intentionally import-path agnostic: we don't require project_dir to be on sys.path and we don't care about the module name outside of this function.

Source code in src/fastflowtransform/hooks/registry.py 
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
def load_project_hooks(project_dir: str | Path) -> None:
    """
    Load all Python files under `<project_dir>/hooks/**.py`.

    This executes the modules (without requiring them to be proper
    Python packages), so any `@fft_hook(...)` calls will populate the
    registry.

    This is intentionally import-path agnostic: we don't require
    `project_dir` to be on sys.path and we don't care about the
    module name outside of this function.
    """
    base = Path(project_dir)
    hooks_dir = base / "hooks"

    if not hooks_dir.is_dir():
        return

    for path in hooks_dir.rglob("*.py"):
        # Build a synthetic, flat module name so we don't rely on package structure
        rel = path.relative_to(base)
        stem_parts = rel.with_suffix("").parts  # e.g. ("hooks", "notify")
        module_name = "_fft_project_hooks_" + "_".join(stem_parts)

        # Skip if already loaded (avoid double-execution)
        if module_name in sys.modules:
            continue

        spec = importlib.util.spec_from_file_location(module_name, path)
        if not spec or not spec.loader:
            continue

        module = importlib.util.module_from_spec(spec)
        sys.modules[module_name] = module
        spec.loader.exec_module(module)