masoora

A helper library for writing data pipelines in Python.

masoora handles the wiring — working out step order, passing data between steps, running independent steps at the same time — so your code stays focused on the transformations.

It works with your orchestrator rather than replacing it. Build a pipeline, package it as a versioned wheel, and call it from an Airflow task: Airflow passes its parameters in as the context, and the pipeline runs as one task. Changing the pipeline is a version bump, not an orchestrator deployment.

• Readable: the builder chain is the pipeline — where data comes from, what happens to it, and where it goes, visible without running anything
• Fluent chaining: .with_read_step() / .with_transform_step() / .with_write_step()
• Pydantic context: typed configuration passed to every step, validated at the boundary
• Data catalog: in-memory key → dataset store (polars/pandas/spark/dicts — anything)
• DAG resolution: declare steps in any order; cycles and missing inputs fail at build()
• Data validation: attach a pandera schema, or any callable, to a catalog key
• Diagrams: pipeline.to_mermaid() renders the DAG, no dependencies
• Parallel: dependency-driven scheduling, no level barriers
• Testable: mock reads/writes and assert on the catalog with a pytest fixture

Installation

pip install masoora

Or with uv:

uv add masoora

Requires Python 3.10+. The only runtime dependency is Pydantic.

The pytest helpers (make_pipeline_fixture) need pytest, available as an extra:

pip install "masoora[pytest]"

Usage

from masoora import PipelineBuilder, PipelineContext


class MyContext(PipelineContext):
    source_url: str
    min_score: float = 0.5


def read_events(ctx: MyContext): ...
def score(ctx: MyContext, events): ...
def filter_top(ctx: MyContext, scored): ...
def write_db(ctx: MyContext, top): ...


pipeline = (
    PipelineBuilder[MyContext]()
    .with_read_step(read_events, output="events")
    .with_transform_step(score, inputs=["events"], output="scored")
    .with_transform_step(filter_top, inputs=["scored"], output="top")
    .with_write_step(write_db, inputs=["top"])
    .build()
)

catalog = pipeline.run(MyContext(source_url="https://..."))

pipeline.to_mermaid() renders the same pipeline as a diagram:

flowchart TD
    n0["read_events"]:::read
    n1["score"]:::transform
    n2["filter_top"]:::transform
    n3["write_db"]:::write
    n0 -->|"events"| n1
    n1 -->|"scored"| n2
    n2 -->|"top"| n3
    classDef read fill:#dbeafe,stroke:#2563eb,color:#0b2a5b;
    classDef transform fill:#e5e7eb,stroke:#4b5563,color:#111827;
    classDef write fill:#dcfce7,stroke:#16a34a,color:#052e16;
    classDef data fill:#fef3c7,stroke:#d97706,color:#451a03;

Step signatures:

Steps may be declared in any order — build() topo-sorts them. Run only what's needed for one output with pipeline.run(ctx, target="top"). Pre-populated catalog keys are declared with .with_seed(key).

Parallel execution

pipeline.run(ctx, parallel=True)  # thread pool, os.cpu_count() workers
pipeline.run(ctx, parallel=4)  # explicit worker count
pipeline.run(ctx, executor=pool)  # your Executor (not shut down by masoora)

Steps run concurrently in a ThreadPoolExecutor with dependency-driven scheduling: each step starts the instant its own dependencies finish — there is no level barrier, so unrelated slow steps never delay a ready branch. Fail-fast: the first step error cancels queued work and raises StepExecutionError immediately; already-running siblings finish in the background.

Contract: steps must only read their declared input keys, write their own output key, and treat the context as read-only. Under this contract parallel results are identical to sequential.

Testing

from masoora import TestRunResult, make_pipeline_fixture

run_pipeline = make_pipeline_fixture(
    pipeline,
    MyContext(source_url="test"),
    reads={"events": fake_events},  # read step is replaced, real source untouched
)


def test_top_events(run_pipeline: TestRunResult[MyContext]) -> None:
    assert run_pipeline.catalog["top"] == expected
    assert run_pipeline.written["top"] == expected  # write step captured, not executed

Or without pytest: pipeline.to_testable(reads={...}).run(ctx) → TestRunResult.

Development

uv sync
uv run pytest
uv run ruff check .
uv run mypy src tests

Issues and pull requests are welcome. The API is still young — if something feels awkward to use, that is worth an issue.

Links

• Documentation
• Changelog
• PyPI
• Issues

License

MIT — see LICENSE.