pipelantic 0.4.0

Typed, contract-driven data pipeline modeling for Python.

Pipelantic

Define, validate, plan, and locally run contract-driven data pipelines with typed Python.

Define data, transformations, and pipelines with typed Python classes. Validate and plan them once. Execute them through interchangeable backends.

Status

0.4.0 — Local Runtime and Operational Model

Pipelantic provides the typed modeling kernel, contract interoperability, an immutable secret-free PipelinePlan, and a local async runtime that executes plans via Python callables, in-memory artifacts, and stdlib JSON/CSV bindings.

Use Pipelantic when you want to catch incompatible wiring before processing data, generate ODCS/DTCS/DPCS contracts from Python, or inspect a deterministic execution plan independently of a future backend.

Pipelantic is currently alpha. Pandas, Polars, SQL, Spark, Airflow, and other external backend plugins are design work and are not included in 0.4.

See the hosted documentation for the full design, CHANGELOG.md for release notes, and Roadmap for sequencing.

Install

pip install pipelantic
# or
uv add pipelantic

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.4.0
git push origin v0.4.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

Capability	0.4
Typed modeling, validation, contracts, and planning	Available
Local Python execution and run reports	Available
Memory, callable, JSON, CSV, and no-write storage	Available
Pandas, Polars, SQL, Spark, and Airflow plugins	Not yet available

Documentation

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

License

MIT