kptn

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.

---

Installation

pip install kptn

Optional 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

---

Concepts

• Tasks — the unit of work. A task is a Python function decorated with @kptn.task, a SQL file registered with kptn.sql_task(), or an R script registered with kptn.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 like parallel(), Stage(), and map() 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 plan to preview what will run before committing.
• Profiles — named configurations in kptn.yaml that filter stages, override task arguments, and control execution cursors. Useful for running a subset of the pipeline for a specific dataset or environment.

---

Quick Start

# 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 run

Or run directly from Python:

kptn.run(pipeline)

---

Core Features

Graph Composition

Chain tasks sequentially with >>:

graph = extract >> transform >> load

Fan out to parallel branches with kptn.parallel():

graph = ingest >> kptn.parallel(transform_a, transform_b) >> merge

Use 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 >> analyze

Use kptn.map() to fan out dynamically over a runtime collection:

graph = list_items >> kptn.map(process_item, over="items")

Profiles

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: false

Profile 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 subset

DuckDB Integration

Pass 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:
    ...

Caching and Re-runs

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 --force

To preview what would run without executing:

kptn plan

---

API Reference

Tasks

Symbol	Description
@kptn.task(outputs, optional=None, compute=None, duckdb_checkpoint=False)	Decorate a Python function as a kptn task
kptn.sql_task(path, outputs, optional=None, duckdb_checkpoint=False)	Register a SQL file as a task
kptn.r_task(path, outputs, compute=None, optional=None, duckdb_checkpoint=False)	Register an R script as a task
kptn.noop()	Placeholder / synchronization node

Graph Composition

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

Pipeline & Execution

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

---

CLI Reference

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