Pipelantic

Catch incompatible data-pipeline wiring before you process data.

Define datasets, transformations, and pipelines as typed Python classes. Validate and plan them once. Run locally today; swap Polars, Pandas, or SQL backends without rewriting the logical pipeline.

Status: Alpha 0.6.0 — local runtime + optional Polars/Pandas/SQL plugins. Spark and Airflow compilation are not shipped.

Install

Requires Python 3.11 or newer.

pip install pipelantic
# optional engines
pip install pipelantic-polars
pip install pipelantic-pandas
pip install pipelantic-sql

Verify:

python -c "import pipelantic; print(pipelantic.__version__)"

Development

Requires uv.

uv sync
uv run pytest
uv run ruff check .
uv run ruff format .

uv sync creates .venv, installs the package in editable mode, and installs the dev dependency group (pytest, ruff, mkdocs) by default.

Release

Tag a version that matches src/pipelantic/_version.py, then push the tag:

git tag v0.6.0
git push origin v0.6.0

GitHub Actions runs checks and publishes to PyPI using the PYPI_API_TOKEN repository secret.

Quick example

from pipelantic import (
    Data,
    Input,
    Output,
    Pipeline,
    PipelineRuntime,
    Sink,
    Source,
    Transformation,
)


class RawCustomer(Data):
    customer_id: int
    first_name: str
    last_name: str


class Customer(Data):
    customer_id: int
    full_name: str


class NormalizeCustomers(Transformation):
    customers: Input[RawCustomer]
    result: Output[Customer]


class CustomerPipeline(Pipeline):
    raw: Source[RawCustomer] = Source(binding="customer_source")
    normalized = NormalizeCustomers.step(customers=raw)
    curated: Sink[Customer] = Sink(
        input=normalized.result,
        binding="customer_sink",
    )


@NormalizeCustomers.implementation("local")
def normalize(customers: list[RawCustomer]) -> list[Customer]:
    return [
        Customer(
            customer_id=row.customer_id,
            full_name=f"{row.first_name} {row.last_name}",
        )
        for row in customers
    ]


CustomerPipeline.validate(profile="development").raise_for_errors()

runtime = PipelineRuntime()
runtime.memory.seed(
    "customer_source",
    [RawCustomer(customer_id=1, first_name="Ada", last_name="Lovelace")],
)
run_report = CustomerPipeline.run(profile="development", runtime=runtime)
print(runtime.memory.get("customer_sink"))

Run the complete tested version at examples/quickstart.py.

Current capability boundary

Documentation

• Documentation site
• Getting Started
• Capabilities and Limitations
• Evaluator brief
• Core Concepts
• Architecture
• Roadmap

License

MIT