masoora

Fluent builder for testable ETL pipelines in Python.

• Fluent chaining: .with_read_step() / .with_transform_step() / .with_write_step()
• Pydantic context: typed configuration passed to every step
• 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()
• Testable: mock reads/writes and assert on the catalog with a pytest fixture

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://..."))

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