Metadata-Version: 2.4
Name: simple_dag
Version: 1.2.1
Summary: Create simple Pipelines with Python
Author-email: Tim Rohner <info@timrohner.ch>
License-Expression: MIT
Project-URL: Source, https://github.com/leokster/simple_dag
Keywords: simple_dag
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: Natural Language :: English
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
Requires-Python: >=3.10
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: aiohttp<4,>=3.9.0
Requires-Dist: click<9,>=8.1
Requires-Dist: dagster-webserver<2,>=1.12.0
Requires-Dist: dagster<2,>=1.12.0
Requires-Dist: fsspec>=2024.2.0
Requires-Dist: matplotlib<4,>=3.8.0
Requires-Dist: pandas<4,>=2.2.0
Requires-Dist: pydantic<3,>=2.0
Requires-Dist: requests<3,>=2.31.0
Provides-Extra: dev
Requires-Dist: black<27,>=24.0.0; extra == "dev"
Requires-Dist: build>=1.0; extra == "dev"
Requires-Dist: flake8>=7.0; extra == "dev"
Requires-Dist: pytest>=8.0; extra == "dev"
Requires-Dist: tox<5,>=4.0; extra == "dev"
Requires-Dist: twine<8,>=6; extra == "dev"
Dynamic: license-file

# simple_dag

![pypi](https://img.shields.io/pypi/v/simple_dag.svg)
[![Documentation Status](https://readthedocs.org/projects/simple-pipeline/badge/?version=latest)](https://simple-pipeline.readthedocs.io/en/latest/?version=latest)
[![Updates](https://pyup.io/repos/github/leokster/simple_dag/shield.svg)](https://pyup.io/repos/github/leokster/simple_dag/)

Orchestration-agnostic Python pipeline library. Decorate functions with `@transform` to define DAG nodes. Run them directly as plain Python callables or integrate with Dagster — no rewriting required.

- Free software: MIT license

## Getting Started

```bash
pip install simple_dag
```

```python
import os
from simple_dag import transform, PandasDFInput, PandasDFOutput

@transform(
    df=PandasDFInput("data/salaries.csv"),
    output=PandasDFOutput("data/salaries_2023.csv"),
)
def filter_2023(df, output: PandasDFOutput):
    df = df[df["work_year"] == 2023]
    output.write_data(df, index=False)

# Run directly
filter_2023()
```

## The Main Ideas

**What is a transform?**
A function decorated with `@transform`. Input arguments are loaded automatically; output arguments are passed through so your function can call `write_data()`.

**Orchestration-agnostic?**
Transforms are plain callables. Call them directly in Python or register them with Dagster using `build_dagster_from_folder`. Switch orchestrators without rewriting pipelines.

## `@transform`

```python
from simple_dag import transform, PandasDFInput, PandasDFOutput

@transform(
    df=PandasDFInput("data/raw.csv"),
    output=PandasDFOutput("data/processed.csv"),
)
def my_step(df, output: PandasDFOutput):
    output.write_data(df[df["value"] > 0], index=False)
```

- Keyword arguments that are `Input` instances are loaded and the result is passed to your function.
- Keyword arguments that are `Output` instances are passed through unchanged so you can call `write_data()`.

## `@schedule`

Stack `@schedule` above `@transform` to attach scheduling metadata:

```python
from simple_dag import transform, schedule, PandasDFInput, PandasDFOutput

@schedule("0 * * * *")              # cron: run every hour
@transform(...)
def hourly_step(...): ...

@schedule(on_upstream_success=True) # trigger when upstream finishes
@transform(...)
def downstream_step(...): ...
```

## Input Types

| Class | Description |
|---|---|
| `PandasDFInput(path, **kwargs)` | Reads a CSV via fsspec; kwargs forwarded to `pd.read_csv`; supports `health_checks` |
| `BinaryInput(path)` | Reads raw bytes; works with any fsspec-compatible path |
| `SparkDFInput(path)` | Spark DataFrame from JSON/Parquet/CSV (experimental) |
| `JsonInput(path, schema_validation=None)` | Reads a JSON file; validates with a Pydantic model if `schema_validation` is set; returns model or raw dict |
| `PathInput(path)` | Pass-through — function receives the raw path string |
| `DirectoryInput(path)` | Validates path is a directory; function receives the path string |
| `Multiple(input_obj)` | Expands a wildcard path into a **list** of loaded inputs |
| `MultipleIterator(input_obj)` | Like `Multiple` but returns a **lazy iterator** (memory-efficient) |

## Output Types

| Class | Description |
|---|---|
| `PandasDFOutput(path)` | Writes a DataFrame to CSV |
| `BinaryOutput(path)` | Writes raw bytes |
| `SparkDFOutput(path)` | Writes a Spark DataFrame to Parquet (experimental) |
| `JsonOutput(path)` | Writes a dict, list, or Pydantic model to JSON |
| `DirectoryOutput(path)` | Holds a path; `write_data` raises `NotImplementedError` — use `.path` directly |

## Health Checks

Pass a list of callables to any input or output. Each receives the loaded data and must return `True` to pass:

```python
def has_salary_col(df):
    return "salary" in df.columns

PandasDFInput("data.csv", health_checks=[has_salary_col])
```

A failed health check raises `ValueError`.

## Direct Execution

Transforms are plain callables — no Dagster required:

```python
filter_2023()   # loads inputs, runs function, writes outputs
```

You can also discover and run all transforms in a folder:

```python
from simple_dag.transforms import find_transform_instances_in_folder

transforms = find_transform_instances_in_folder("my_pipeline/")
for t in transforms:
    t()
```

## Dagster Integration

```python
# definitions.py
from simple_dag.orchestrators.dagster import build_definitions

defs = build_definitions("my_project.pipelines")  # package name, module, folder or list
```

`build_definitions` turns every transform into an asset (grouped by the first folder of a
base-relative path), every `@schedule(cron=...)` into a `ScheduleDefinition` that is running
from the start, and `on_upstream_success` transforms into an automation-condition sensor.
Pass `running=False` to create schedules in the stopped state.

Launch the Dagster UI:

```bash
dagster dev -m my_project.definitions
```

The older `build_dagster_from_folder(folder)` / `build_dagster(transforms)` helpers still
return `(assets, source_assets)` for hand-built `Definitions`.

## Cloud Storage

All path arguments accept fsspec-compatible URIs — no code changes needed:

- `s3://my-bucket/path/to/data.csv`
- `abfs://my-container/path/to/data.csv` (Azure Blob)
- `/local/path/to/data.csv`

## Base Path

Most pipelines read and write below one root. Configure it once and use relative paths in
the handlers; the same code then runs against a local folder on a laptop and against a
bucket or an Azure container in production:

```bash
export SIMPLE_DAG_BASE=~/data                                   # laptop
export SIMPLE_DAG_BASE=abfs://data@account.dfs.core.windows.net  # production
```

```python
from simple_dag import JsonInput, JsonOutput, transform

@transform(
    orders=JsonInput("sales/raw/orders.json"),
    out=JsonOutput("sales/curated/orders.json"),
)
def clean_orders(orders, out):
    out.write_data([o for o in orders if o["amount"] > 0])
```

- `simple_dag.configure(base=..., storage_options=...)` does the same from code;
  `SIMPLE_DAG_STORAGE_OPTIONS` takes a JSON object of fsspec options for the base.
- Below the base, Azure is accessed authenticated (`anon=False`, i.e. `DefaultAzureCredential`:
  workload identity, managed identity or `az login`) and local writes create missing parent
  directories. Absolute paths and URIs behave as before.
- Relative paths also give stable, base-independent asset names in Dagster
  (`sales/curated/orders.json` → `sales_curated_orders_json`, group `sales`).
- `simple_dag.utils.fsspec.open / exists / remove / list_files` apply the same resolution for
  code outside the handlers.

## Examples

### JSON + Pydantic

```python
from pydantic import BaseModel
from simple_dag import transform, JsonInput, JsonOutput

class SensorReading(BaseModel):
    sensor_id: str
    value: float

@transform(
    reading=JsonInput("data/reading.json", schema_validation=SensorReading),
    output=JsonOutput("data/alert.json"),
)
def evaluate(reading: SensorReading, output: JsonOutput):
    result = {"sensor_id": reading.sensor_id, "triggered": reading.value > 100}
    output.write_data(result)
```

### Multiple files with wildcard

```python
from simple_dag import transform, Multiple, PandasDFInput, PandasDFOutput

@transform(
    dfs=Multiple(PandasDFInput("data/raw/*.csv")),
    output=PandasDFOutput("data/combined.csv"),
)
def combine(dfs, output: PandasDFOutput):
    import pandas as pd
    output.write_data(pd.concat(dfs), index=False)
```

See `examples/ds_salaries/` for a full multi-step pipeline including schedules, health checks, and a plot output.
