Run pipelines from the command line

The data-joinery command runs a pipeline defined in a Python file. It can also draw the pipeline, so you can check its steps without running them.

Define a pipeline

Create a pipeline object or a build_pipeline() function that returns one. The CLI loads that pipeline when you run a command. This example builds a report for a date supplied at run time:

from dataclasses import dataclass
from datetime import date
from typing import Annotated

from data_joinery import Context, Pipeline, side_effect, transform


class RunDate(date):
    pass


@dataclass(frozen=True)
class PipelineContext:
    run_date: RunDate

    @classmethod
    def from_kwargs(cls, *, run_date: str) -> "PipelineContext":
        return cls(run_date=RunDate.fromisoformat(run_date))


@transform
def make_report(run_date: Annotated[RunDate, Context()]) -> str:
    return f"Report for {run_date.isoformat()}"


@side_effect
def print_report(report: str) -> None:
    print(report)


def build_pipeline() -> Pipeline[PipelineContext]:
    pipeline = Pipeline(PipelineContext)
    pipeline.add_step(make_report) >> pipeline.add_step(print_report)
    return pipeline


if __name__ == "__main__":
    build_pipeline().run(PipelineContext.from_kwargs(run_date="2026-10-11"))

Outputs:

Report for 2026-10-11

The pipeline has a PipelineContext dataclass because make_report needs a run date. The Context() marker supplies that value to the transform. See Context for more about passing values to steps.

Run it with a date

Pass the path to the Python file and a value for run_date:

data-joinery run docs_src/learn/pipelines/cli_from_kwargs.py --context run_date=2026-10-11

The command prints Report for 2026-10-11. Each --context NAME=VALUE becomes a string keyword argument to PipelineContext.from_kwargs(). In this example, from_kwargs converts the string to a RunDate. Add another --context option for each additional value your context needs. A context class must define from_kwargs to accept values from the command line.

If your pipeline has no context, leave out --context. When a context needs no command line values, the CLI calls from_kwargs() if it exists, or creates the context dataclass with no arguments.

Run without side effects

The example marks print_report with @side_effect. To run make_report without printing the report, pass --skip-side-effects:

data-joinery run docs_src/learn/pipelines/cli_from_kwargs.py --context run_date=2026-10-11 --skip-side-effects

The report is still made, but print_report is skipped. Other steps that are not marked as side effects continue to run. See Side effects for when to mark a step this way.

Draw the pipeline

Install data-joinery[vis] to use visualize. This command opens a Matplotlib window showing the pipeline's steps:

data-joinery visualize docs_src/learn/pipelines/cli_from_kwargs.py

You can save the drawing as an SVG or PNG file instead of opening a window:

data-joinery visualize docs_src/learn/pipelines/cli_from_kwargs.py --output report.svg
data-joinery visualize docs_src/learn/pipelines/cli_from_kwargs.py --output report.png

Choose a pipeline file or module

If you leave out the path, the CLI looks for pipeline or build_pipeline() in main.py. For another file or module, pass its path or import name. Add :name when the pipeline has a different name:

data-joinery run
data-joinery run pipelines/orders.py:orders_pipeline
data-joinery run my_project.orders:build_pipeline