API reference

Everything exported from the top-level masoora package. Generated from the source, so it always matches the installed version.

Building

masoora.PipelineBuilder

PipelineBuilder()

Bases: Generic[ContextT]

Chain read/transform/write steps, then build() a validated Pipeline.

Steps may be declared in any order; build() resolves dependencies (DAG topo-sort) and fails fast on unknown inputs, duplicate outputs, or cycles.

Source code in src/masoora/builder.py

def __init__(self) -> None:
    self._steps: list[Step[ContextT]] = []
    self._seeds: set[str] = set()
    self._validators: dict[str, Validator] = {}

build

build() -> Pipeline[ContextT]

Validate and topo-sort the steps into an executable Pipeline.

Source code in src/masoora/builder.py

def build(self) -> Pipeline[ContextT]:
    """Validate and topo-sort the steps into an executable Pipeline."""
    self._validate()
    return Pipeline(steps=topo_sort(self._steps), validators=self._validators)

with_read_step

with_read_step(
    fn: Callable[[ContextT], Any],
    *,
    output: str,
    validate: Validator | None = None,
) -> Self

fn(context) -> dataset, stored at catalog[output].

Source code in src/masoora/builder.py

def with_read_step(
    self, fn: Callable[[ContextT], Any], *, output: str, validate: Validator | None = None
) -> Self:
    """fn(context) -> dataset, stored at catalog[output]."""
    self._steps.append(ReadStep(fn=fn, output=output))
    self._register_validator(output, validate)
    return self

with_seed

with_seed(
    key: str, *, validate: Validator | None = None
) -> Self

Declare a catalog key that will be pre-populated before run().

Source code in src/masoora/builder.py

def with_seed(self, key: str, *, validate: Validator | None = None) -> Self:
    """Declare a catalog key that will be pre-populated before run()."""
    self._seeds.add(key)
    self._register_validator(key, validate)
    return self

with_transform_step

with_transform_step(
    fn: Callable[..., Any],
    *,
    inputs: Sequence[str],
    output: str,
    validate: Validator | None = None,
) -> Self

fn(context, *inputs) -> dataset, stored at catalog[output].

Source code in src/masoora/builder.py

def with_transform_step(
    self,
    fn: Callable[..., Any],
    *,
    inputs: Sequence[str],
    output: str,
    validate: Validator | None = None,
) -> Self:
    """fn(context, *inputs) -> dataset, stored at catalog[output]."""
    self._steps.append(TransformStep(fn=fn, inputs=inputs, output=output))
    self._register_validator(output, validate)
    return self

with_write_step

with_write_step(
    fn: Callable[..., None], *, inputs: Sequence[str]
) -> Self

fn(context, *inputs) -> None; terminal step.

Source code in src/masoora/builder.py

def with_write_step(self, fn: Callable[..., None], *, inputs: Sequence[str]) -> Self:
    """fn(context, *inputs) -> None; terminal step."""
    self._steps.append(WriteStep(fn=fn, inputs=inputs))
    return self

masoora.PipelineContext

Bases: BaseModel

Holds all variables/configuration a pipeline needs.

Users subclass this and declare their own fields:

class MyContext(PipelineContext):
    source_url: str
    batch_size: int = 100

Running

masoora.Pipeline

Pipeline(
    steps: Sequence[Step[ContextT]],
    validators: Mapping[str, Validator] | None = None,
)

Bases: Generic[ContextT]

A topo-sorted sequence of steps. Built via PipelineBuilder.

Source code in src/masoora/pipeline.py

def __init__(
    self,
    steps: Sequence[Step[ContextT]],
    validators: Mapping[str, Validator] | None = None,
) -> None:
    self._steps: tuple[Step[ContextT], ...] = tuple(steps)
    self._validators: dict[str, Validator] = dict(validators) if validators else {}

validators property

validators: Mapping[str, Validator]

Catalog key -> validator, as declared on the builder.

run

run(
    context: ContextT,
    catalog: DataCatalog | None = None,
    target: str | None = None,
    parallel: ParallelMode = False,
    executor: Executor | None = None,
) -> DataCatalog

Execute all steps (or only those needed for target) in topo order.

parallel: False (default) runs sequentially; True uses a thread pool with os.cpu_count() workers; an int sets the worker count. Steps run concurrently, each starting the instant its own dependencies finish (dependency-driven scheduling, no level barrier). executor: caller-provided Executor; wins over parallel and is NOT shut down by the pipeline.

Source code in src/masoora/pipeline.py

def run(
    self,
    context: ContextT,
    catalog: DataCatalog | None = None,
    target: str | None = None,
    parallel: ParallelMode = False,
    executor: Executor | None = None,
) -> DataCatalog:
    """Execute all steps (or only those needed for `target`) in topo order.

    parallel: False (default) runs sequentially; True uses a thread pool
    with os.cpu_count() workers; an int sets the worker count. Steps run
    concurrently, each starting the instant its own dependencies finish
    (dependency-driven scheduling, no level barrier).
    executor: caller-provided Executor; wins over `parallel` and is NOT
    shut down by the pipeline.
    """
    steps = self._select(target)
    cat = catalog if catalog is not None else DataCatalog()
    # A caller-supplied catalog holds the seeded keys, so it needs the
    # validators too -- that is what gets seeds checked on first read.
    cat.attach_validators(self._validators)
    _dispatch(steps, context, cat, parallel, executor)
    return cat

to_mermaid

to_mermaid(
    *, target: str | None = None, direction: str = "TD"
) -> str

Render the pipeline as a Mermaid flowchart definition.

Steps are nodes and catalog keys are edge labels; seeded inputs and unconsumed outputs get their own nodes. Paste the result into a ```mermaid fence -- GitHub, MkDocs Material and Jupyter all render it.

target: diagram only the steps needed to produce that key. direction: any Mermaid flowchart direction (TD, LR, ...).

Source code in src/masoora/pipeline.py

def to_mermaid(self, *, target: str | None = None, direction: str = "TD") -> str:
    """Render the pipeline as a Mermaid flowchart definition.

    Steps are nodes and catalog keys are edge labels; seeded inputs and
    unconsumed outputs get their own nodes. Paste the result into a
    ```mermaid fence -- GitHub, MkDocs Material and Jupyter all render it.

    target: diagram only the steps needed to produce that key.
    direction: any Mermaid flowchart direction (TD, LR, ...).
    """
    return to_mermaid(self._select(target), direction=direction)

to_testable

to_testable(
    reads: Mapping[str, Any] | None = None,
    *,
    mock_writes: bool = True,
    target: str | None = None,
) -> TestablePipeline[ContextT]

Return a copy with read steps replaced by fixture data and, optionally, write steps captured instead of executed.

Source code in src/masoora/pipeline.py

def to_testable(
    self,
    reads: Mapping[str, Any] | None = None,
    *,
    mock_writes: bool = True,
    target: str | None = None,
) -> TestablePipeline[ContextT]:
    """Return a copy with read steps replaced by fixture data and,
    optionally, write steps captured instead of executed."""
    read_fixtures = dict(reads) if reads else {}
    mocked: list[Step[ContextT]] = []
    for step in self._select(target):
        if isinstance(step, ReadStep) and step.output in read_fixtures:
            value = read_fixtures[step.output]

            def constant_read(_ctx: ContextT, v: Any = value) -> Any:
                return v

            mocked.append(ReadStep(fn=constant_read, output=step.output))
        elif isinstance(step, WriteStep) and mock_writes:
            mocked.append(step)  # capture happens in TestablePipeline.run
        else:
            mocked.append(step)
    return TestablePipeline(steps=mocked, mock_writes=mock_writes, validators=self._validators)

masoora.DataCatalog

DataCatalog(
    initial: Mapping[str, Any] | None = None,
    validators: Mapping[str, Validator] | None = None,
)

Bases: MutableMapping[str, Any]

Holds datasets produced and consumed by pipeline steps.

Values are unconstrained: polars/pandas/spark dataframes, dicts, models, etc.

A key may carry a validator. Values are checked when written and when read, so data that never passed through a step -- seeded keys, fixtures supplied by to_testable() -- is checked too. Each value is checked once: a read after a validated write does not repeat the work.

Source code in src/masoora/catalog.py

def __init__(
    self,
    initial: Mapping[str, Any] | None = None,
    validators: Mapping[str, Validator] | None = None,
) -> None:
    self._data: dict[str, Any] = dict(initial) if initial else {}
    self._validators: dict[str, Validator] = dict(validators) if validators else {}
    self._validated: set[str] = set()

attach_validators

attach_validators(
    validators: Mapping[str, Validator],
) -> None

Register validators for keys that do not already have one.

Source code in src/masoora/catalog.py

def attach_validators(self, validators: Mapping[str, Validator]) -> None:
    """Register validators for keys that do not already have one."""
    for key, validator in validators.items():
        self._validators.setdefault(key, validator)

snapshot

snapshot() -> dict[str, Any]

Return a shallow copy of the catalog contents, skipping validation.

Source code in src/masoora/catalog.py

def snapshot(self) -> dict[str, Any]:
    """Return a shallow copy of the catalog contents, skipping validation."""
    return dict(self._data)

Validation

masoora.Validator module-attribute

Validator = SupportsValidate | Callable[[Any], Any]

Steps

masoora.ReadStep dataclass

ReadStep(fn: Callable[[ContextT], Any], output: str)

Bases: Generic[ContextT]

fn(context) -> dataset, stored at catalog[output].

masoora.TransformStep dataclass

TransformStep(
    fn: Callable[..., Any],
    inputs: Sequence[str],
    output: str,
)

Bases: Generic[ContextT]

fn(context, *inputs) -> dataset, stored at catalog[output].

Source code in src/masoora/steps.py

def __init__(self, fn: Callable[..., Any], inputs: Sequence[str], output: str) -> None:
    object.__setattr__(self, "fn", fn)
    object.__setattr__(self, "inputs", tuple(inputs))
    object.__setattr__(self, "output", output)

masoora.WriteStep dataclass

WriteStep(fn: Callable[..., None], inputs: Sequence[str])

Bases: Generic[ContextT]

fn(context, *inputs) -> None; produces no catalog output.

Source code in src/masoora/steps.py

def __init__(self, fn: Callable[..., None], inputs: Sequence[str]) -> None:
    object.__setattr__(self, "fn", fn)
    object.__setattr__(self, "inputs", tuple(inputs))

Testing

masoora.TestablePipeline

TestablePipeline(
    steps: Sequence[Step[ContextT]],
    *,
    mock_writes: bool,
    validators: Mapping[str, Validator] | None = None,
)

Bases: Generic[ContextT]

A pipeline with mocked reads/writes; run() returns a TestRunResult.

Source code in src/masoora/pipeline.py

def __init__(
    self,
    steps: Sequence[Step[ContextT]],
    *,
    mock_writes: bool,
    validators: Mapping[str, Validator] | None = None,
) -> None:
    self._steps: tuple[Step[ContextT], ...] = tuple(steps)
    self._mock_writes = mock_writes
    self._validators: dict[str, Validator] = dict(validators) if validators else {}

masoora.TestRunResult dataclass

TestRunResult(
    catalog: DataCatalog, written: dict[str, Any] = dict()
)

Bases: Generic[ContextT]

Result of running a TestablePipeline.

catalog: full data catalog after the run (assert on any key). written: input key -> dataset for every mocked write step.

masoora.make_pipeline_fixture

make_pipeline_fixture(
    pipeline: Pipeline[ContextT],
    context: ContextT | Callable[[], ContextT],
    reads: Mapping[str, Any] | None = None,
    *,
    mock_writes: bool = True,
    target: str | None = None,
) -> Callable[[], TestRunResult[ContextT]]

Create a pytest fixture that runs the pipeline with mocked IO.

Usage

run_pipeline = make_pipeline_fixture( my_pipeline, MyContext(url="test"), reads={"raw": fake_df} )

def test_output(run_pipeline: TestRunResult[MyContext]) -> None: assert "clean" in run_pipeline.catalog

Source code in src/masoora/testing.py

def make_pipeline_fixture(
    pipeline: Pipeline[ContextT],
    context: ContextT | Callable[[], ContextT],
    reads: Mapping[str, Any] | None = None,
    *,
    mock_writes: bool = True,
    target: str | None = None,
) -> Callable[[], TestRunResult[ContextT]]:
    """Create a pytest fixture that runs the pipeline with mocked IO.

    Usage:
        run_pipeline = make_pipeline_fixture(
            my_pipeline, MyContext(url="test"), reads={"raw": fake_df}
        )

        def test_output(run_pipeline: TestRunResult[MyContext]) -> None:
            assert "clean" in run_pipeline.catalog
    """
    try:
        import pytest
    except ImportError as exc:  # pragma: no cover
        raise ImportError("make_pipeline_fixture requires pytest; install masoora[pytest]") from exc

    @pytest.fixture
    def _fixture() -> TestRunResult[ContextT]:
        ctx = context() if callable(context) else context
        return pipeline.to_testable(reads=reads, mock_writes=mock_writes, target=target).run(ctx)

    return _fixture

Errors

masoora.PipelineError

Bases: Exception

Base class for all masoora errors.

masoora.PipelineValidationError

Bases: PipelineError

Raised at build time when the pipeline definition is invalid.

masoora.PipelineCycleError

PipelineCycleError(cycle: list[str])

Bases: PipelineValidationError

Raised at build time when steps form a dependency cycle.

Source code in src/masoora/errors.py

def __init__(self, cycle: list[str]) -> None:
    self.cycle = cycle
    super().__init__(f"Pipeline contains a dependency cycle: {' -> '.join(cycle)}")

masoora.DataValidationError

DataValidationError(
    key: str, phase: str, original: Exception
)

Bases: PipelineError

Raised when a catalog value fails the validator declared for its key.

Source code in src/masoora/errors.py

def __init__(self, key: str, phase: str, original: Exception) -> None:
    self.key = key
    self.phase = phase
    self.original = original
    super().__init__(
        f"Data for catalog key {key!r} failed validation on {phase}: "
        f"{type(original).__name__}: {original}"
    )

masoora.StepExecutionError

StepExecutionError(
    step_index: int, step_name: str, original: Exception
)

Bases: PipelineError

Raised when a step fails during run().

Source code in src/masoora/errors.py

def __init__(self, step_index: int, step_name: str, original: Exception) -> None:
    self.step_index = step_index
    self.step_name = step_name
    self.original = original
    super().__init__(
        f"Step {step_index} ({step_name!r}) failed: {type(original).__name__}: {original}"
    )