Metadata-Version: 2.4
Name: masoora
Version: 0.2.0
Summary: Helper library for writing readable, testable data pipelines
Keywords: etl,pipeline,dag,data-engineering,builder,testing
Author: Ahmed Osman
Author-email: Ahmed Osman <79141373+ahmedtilal@users.noreply.github.com>
License-Expression: MIT
License-File: LICENSE
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Typing :: Typed
Requires-Dist: pydantic>=2.13.4
Requires-Dist: pandera>=0.32.1 ; extra == 'pandera'
Requires-Dist: pytest>=9.0 ; extra == 'pytest'
Requires-Python: >=3.10
Project-URL: Changelog, https://github.com/ahmedtilal/masoora/blob/master/CHANGELOG.md
Project-URL: Documentation, https://ahmedtilal.github.io/masoora/
Project-URL: Homepage, https://ahmedtilal.github.io/masoora/
Project-URL: Issues, https://github.com/ahmedtilal/masoora/issues
Project-URL: Repository, https://github.com/ahmedtilal/masoora
Provides-Extra: pandera
Provides-Extra: pytest
Description-Content-Type: text/markdown

# masoora

[![CI](https://github.com/ahmedtilal/masoora/actions/workflows/ci.yml/badge.svg)](https://github.com/ahmedtilal/masoora/actions/workflows/ci.yml)
[![PyPI](https://img.shields.io/pypi/v/masoora.svg)](https://pypi.org/project/masoora/)
[![Python versions](https://img.shields.io/pypi/pyversions/masoora.svg)](https://pypi.org/project/masoora/)
[![License: MIT](https://img.shields.io/badge/license-MIT-blue.svg)](https://github.com/ahmedtilal/masoora/blob/master/LICENSE)
[![Typed](https://img.shields.io/badge/typing-strict-blue.svg)](https://peps.python.org/pep-0561/)

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
- **Parallel**: dependency-driven scheduling, no level barriers
- **Testable**: mock reads/writes and assert on the catalog with a pytest fixture

## Installation

```bash
pip install masoora
```

Or with [uv](https://docs.astral.sh/uv/):

```bash
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:

```bash
pip install "masoora[pytest]"
```

## Usage

```python
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:

| Step kind | Signature | Effect |
|---|---|---|
| read | `fn(ctx) -> dataset` | `catalog[output] = result` |
| transform | `fn(ctx, *inputs) -> dataset` | `catalog[output] = result` |
| write | `fn(ctx, *inputs) -> None` | terminal |

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

```python
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

```python
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

```bash
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](https://ahmedtilal.github.io/masoora/)
- [Changelog](https://github.com/ahmedtilal/masoora/blob/master/CHANGELOG.md)
- [PyPI](https://pypi.org/project/masoora/)
- [Issues](https://github.com/ahmedtilal/masoora/issues)

## License

MIT — see [LICENSE](https://github.com/ahmedtilal/masoora/blob/master/LICENSE).
