Lightweight, cacheable data pipelines in Python, R, and SQL.
kptn lets you define data pipelines as composable Python functions, SQL files, and R scripts. Pipelines are hash-cached by default — unchanged tasks are skipped automatically on re-runs, so you only pay for what changed. Profiles let you parameterize and filter runs by environment or dataset without changing code. Cloud deployment to AWS is on the roadmap.
pip install kptnOptional extras:
pip install kptn[duckdb] # DuckDB state store and SQL tasks
pip install kptn[aws] # AWS deployment (work in progress)CLI setup — add to your project's pyproject.toml:
[tool.kptn]
pipeline = "your_package.pipeline" # module that exposes a `pipeline` attribute- Tasks — the unit of work. A task is a Python function decorated with
@kptn.task, a SQL file registered withkptn.sql_task(), or an R script registered withkptn.r_task(). Each task declares its output files. - Graphs — tasks are composed into a directed acyclic graph using the
>>operator for sequential chaining and operators likeparallel(),Stage(), andmap()for branching. - Caching — kptn hashes each task's outputs and source code. On re-run, tasks whose outputs and dependencies haven't changed are skipped. Use
kptn planto preview what will run before committing. - Profiles — named configurations in
kptn.yamlthat filter stages, override task arguments, and control execution cursors. Useful for running a subset of the pipeline for a specific dataset or environment.
# pipeline.py
import kptn
from pathlib import Path
def get_greeting() -> str:
return "Hello, kptn!"
@kptn.task(outputs=["output/extract.txt"])
def extract(greeting: str) -> None:
Path("output").mkdir(exist_ok=True)
Path("output/extract.txt").write_text(greeting)
@kptn.task(outputs=["output/transform.txt"])
def transform() -> None:
data = Path("output/extract.txt").read_text()
Path("output/transform.txt").write_text(data.upper())
@kptn.task(outputs=["output/load.txt"])
def load() -> None:
data = Path("output/transform.txt").read_text()
Path("output/load.txt").write_text(f"Loaded: {data}")
deps = kptn.config(greeting=get_greeting)
graph = deps >> extract >> transform >> load
pipeline = kptn.Pipeline("hello_kptn", graph)Preview the plan, then run:
kptn plan
kptn runOr run directly from Python:
kptn.run(pipeline)Chain tasks sequentially with >>:
graph = extract >> transform >> loadFan out to parallel branches with kptn.parallel():
graph = ingest >> kptn.parallel(transform_a, transform_b) >> mergeUse kptn.Stage() to define profile-selectable branches. The profile controls which branches are active at runtime:
datasets = kptn.Stage(
"datasets",
load_full,
load_subset,
)
graph = ingest >> datasets >> analyzeUse kptn.map() to fan out dynamically over a runtime collection:
graph = list_items >> kptn.map(process_item, over="items")Declare prerequisites with requires=[...] instead of chaining them with >>. A required task is pulled into the run only when a consumer needs it, runs once even if several consumers require it, and is ordered before every requirer. This is ideal for expensive shared prerequisites:
@kptn.task(outputs=["duckdb://index"])
def build_index(engine) -> None:
... # expensive; only worth running when something needs the index
@kptn.task(outputs=["output/report.txt"], requires=[build_index])
def report(engine) -> None:
...
# build_index is never chained with >> — `report` pulls it in:
pipeline = kptn.Pipeline("analysis", report)requires is transitive: a required task's own requires are pulled in too. If you already place a task in the graph yourself (via >>), requires for that task is a no-op — your explicit wiring governs ordering. Note that conjunctive requires only injects and orders prerequisites — it never drops a consumer. If a user-placed prerequisite is later pruned by a profile, its consumer still runs; use kptn.any_of(...) when you need a missing prerequisite to skip the consumer.
Use kptn.any_of(...) for a disjunctive requirement — a gate that pulls nothing and is satisfied only if one of its members is already in the run. If none is present, the consumer is skipped:
@kptn.task(
outputs=["output/combined.txt"],
requires=[kptn.any_of(load_full, load_subset)],
)
def summarize(engine) -> None:
...This pairs naturally with kptn.Stage(): whichever branch the active profile selects satisfies the any_of gate, and summarize runs against it.
Profiles are defined in kptn.yaml at your project root. They let you parameterize runs without changing code.
settings:
db: duckdb
db_path: pipeline.db
profiles:
full:
stage_selections:
datasets: [load_full]
subset:
stage_selections:
datasets: [load_subset]
args:
analyze:
limit: 1000
subset_test:
extends: subset
stop_after: transform
optional_groups:
qa_checks: falseProfile keys:
| Key | Description |
|---|---|
extends |
Inherit settings from another profile (or a list of profiles) |
stage_selections |
Map of stage name → list of branch names to activate |
args |
Per-task keyword argument overrides |
start_from |
Skip tasks before this task name |
stop_after |
Skip tasks after this task name |
optional_groups |
Enable or disable named optional task groups |
Run with a profile:
kptn run --profile subset
kptn plan --profile subsetPass a DuckDB connection factory via kptn.config(). Tasks receive the connection as a keyword argument:
import duckdb
import kptn
from pathlib import Path
def get_engine():
return duckdb.connect("pipeline.db")
@kptn.task(outputs=["output/summary.parquet"])
def summarize(engine) -> None:
engine.execute("COPY (SELECT * FROM raw) TO 'output/summary.parquet'")
ingest = kptn.sql_task("sql/ingest.sql", outputs=["raw"])
deps = kptn.config(duckdb=(get_engine, "engine"))
graph = deps >> ingest >> summarize
pipeline = kptn.Pipeline("my_pipeline", graph)The duckdb=(factory, "alias") tuple tells kptn to inject the connection under the name "engine". SQL tasks receive it automatically.
Add duckdb_checkpoint=True to a task to persist a DuckDB checkpoint after it runs, enabling incremental restores:
@kptn.task(outputs=["output/final.parquet"], duckdb_checkpoint=True)
def finalize(engine) -> None:
...kptn hashes each task's declared outputs and source code. On re-run:
- Tasks whose outputs exist and haven't changed are skipped.
- Tasks whose source code or upstream dependencies changed are re-run.
To bypass the cache for a single run:
kptn run --forceTo preview what would run without executing:
kptn plan| Symbol | Description |
|---|---|
@kptn.task(outputs, optional=None, compute=None, duckdb_checkpoint=False, requires=None) |
Decorate a Python function as a kptn task |
kptn.sql_task(path, outputs, optional=None, duckdb_checkpoint=False, requires=None) |
Register a SQL file as a task |
kptn.r_task(path, outputs, compute=None, optional=None, duckdb_checkpoint=False, requires=None) |
Register an R script as a task |
kptn.noop() |
Placeholder / synchronization node |
| Symbol | Description |
|---|---|
>> |
Chain tasks or graphs sequentially |
kptn.parallel(*branches) |
Fan out to parallel branches. Accepts an optional name as the first argument: kptn.parallel("name", a, b) |
kptn.Stage(name, *branches) |
Profile-selectable branches grouped under a named stage |
kptn.map(task_fn, over="key") |
Dynamic fanout over a runtime collection |
kptn.any_of(*tasks) |
Disjunctive requirement group for requires= — satisfied if any member task is present in the run (otherwise the consumer is skipped) |
| Symbol | Description |
|---|---|
kptn.config(**kwargs) |
Declare dependency injection factories. Use duckdb=(factory, "alias") for DuckDB connections |
kptn.Pipeline(name, graph) |
Wrap a graph in a named pipeline |
kptn.run(pipeline, *, profile=None, keep_db_open=False, no_cache=False, force=False) |
Execute the pipeline |
kptn.plan(pipeline, *, profile=None) |
Dry-run: print which tasks would run or be skipped |
The kptn CLI discovers your pipeline from [tool.kptn] pipeline = "..." in pyproject.toml. The referenced module must expose a pipeline attribute of type Pipeline.
kptn plan [--profile PROFILE] # preview what will run or be skipped
kptn run [--profile PROFILE] [--force] # execute the pipeline