Metadata-Version: 2.4
Name: masoora
Version: 0.1.0
Summary: Fluent builder for testable ETL 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: pytest>=9.0 ; extra == 'pytest'
Requires-Python: >=3.10
Project-URL: Homepage, https://github.com/ahmedtilal/masoora
Project-URL: Issues, https://github.com/ahmedtilal/masoora/issues
Project-URL: Repository, https://github.com/ahmedtilal/masoora
Provides-Extra: pytest
Description-Content-Type: text/markdown

# 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

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