kedro-intercept

A Kedro hook for attaching interceptors to pipeline lifecycle phases. Unlike writing a hook class, interceptors are declared in the pipeline definition — scoped to the node or dataset they belong to.

Full documentation at saemeon.github.io/kedro-intercept

Installation

pip install kedro-intercept

The hook is auto-discovered via Kedro's entry point mechanism. No settings.py changes needed.

Usage

Call intercept_node next to the node it applies to:

from kedro_intercept import intercept_node

my_node = node(func=train, inputs="raw", outputs="model", name="train_model")

intercept_node(
    my_node,
    before_node_run=log_start,
    after_dataset_saved={"model": validate_model},
)

For global phases, node can be omitted:

intercept_node(before_pipeline_run=setup, after_pipeline_run=teardown)

Or use intercept directly by phase name:

from kedro_intercept import intercept

intercept("before_pipeline_run", setup)
intercept("before_node_run", log_start, name="train_model")
intercept("after_dataset_saved", validate_model, name="model")

Parameter matching

Interceptors only need to declare the parameters they use. Everything else is dropped:

def log_start(node, run_id):
    print(f"{node.name} started — run {run_id}")

Available parameters depend on the phase. See the Kedro hooks documentation for the full signatures.

Phases

Global

Node-level

Dataset-level

Dataset phases accept a single callable (applied to all inputs/outputs), a list, or a dict keyed by dataset name:

intercept_node(
    my_node,
    after_dataset_saved={"predictions": validate, "metrics": log_metrics},
)

Assert variants

Every phase has an assert_ variant that re-raises on failure instead of logging a warning:

intercept_node(
    my_node,
    assert_after_dataset_saved={"predictions": check_no_nulls},
)

Logging

Kedro's default logging only covers the kedro namespace. To see output from this package, add to conf/logging.yml:

loggers:
  kedro_intercept:
    level: INFO

License

MIT