Metadata-Version: 2.3
Name: banzai-pipe
Version: 0.1.0
Summary: Lightweight functional data pipelines backed by files
Requires-Dist: loguru>=0.7.3
Requires-Dist: polars>=1.44.1
Requires-Dist: pyarrow>=25.0.1
Requires-Dist: joblib>=1.6.0 ; extra == 'joblib'
Requires-Dist: obstore>=0.11.1 ; extra == 's3'
Requires-Python: >=3.13
Provides-Extra: joblib
Provides-Extra: s3
Description-Content-Type: text/markdown

# banzai

Lightweight package for functional data pipelines.

A piece of work is done when its output file exists. Everything else is bookkeeping around that.

- A **Unit** is one piece of work's name: hive keys plus, for raw files, the file name. `Unit(symbol="AAPL", day="2026-09-01", file="prices.csv")`.
- A **Dataset** is a name in a store. It owns the path recipe: `<name>/<keys>/<file>`, relative to the store.
- A **Feed** is your function that decides which units need doing right now. It returns a list of units.
- A **Job** is your function that does one unit. It returns bytes or a polars frame. `job.run(unit)` writes that at the unit's path, overwriting. It never checks first. That's the feed's job.
- **tick()** calls each job's feed and runs the job on every unit that comes back.

Nothing tracks state. Status is a listing of the store.

## Install

```sh
uv add banzai-pipe            # local stores, polars frames
uv add banzai-pipe[s3]        # plus S3 via obstore
uv add banzai-pipe[joblib]    # plus a multi-core runner
```

## Getting started

### 1. Pull daily price files from an API

One CSV per symbol per day. The source is immutable, so a unit is done once its file exists.

```python
import datetime as dt
import os
import urllib.request

import banzai
from banzai import stores

SYMBOLS = ["AAPL", "MSFT", "NVDA"]
raw = banzai.Dataset("prices", stores.LocalStore("data/raw"))


@banzai.feed
def recent_days():
    # the last 7 days for every symbol, minus the ones already in the store
    today = dt.date.today()
    units = [
        banzai.Unit(symbol=s, day=today - dt.timedelta(days=i), file="prices.csv")
        for i in range(7)
        for s in SYMBOLS
    ]
    return [u for u in units if u not in raw.have(units)]


@banzai.job(raw, feed=recent_days)
def pull(unit):
    url = f"https://example.com/v1/{unit.symbol}/{unit.day}.csv"
    return urllib.request.urlopen(url).read()


if __name__ == "__main__":
    banzai.tick([pull])
```

Run it once and the store looks like this:

```text
data/raw/prices/symbol=AAPL/day=2026-09-01/prices.csv
data/raw/prices/symbol=AAPL/day=2026-08-31/prices.csv
...
```

Run it again and nothing happens, because the feed sees every file is there. Delete one and the next run redoes exactly that one. Put it on a cron and it keeps the last week filled in. `raw.have(units)` is one listing, not one request per file, so it stays cheap at ten thousand units.

The path is the whole record. `pull(unit)` on its own just returns bytes, so you can call it in a notebook with no framework in the way.

### 2. Parse them into parquet

A second job reads the raw files and writes frames. Its feed is "raw units that have no parsed counterpart". Returning a polars frame makes the leaf `<name>.parquet`, written by polars' own writer.

```python
import io

import polars as pl

parsed = banzai.Dataset("prices", stores.LocalStore("data/parsed"))


@banzai.feed
def unparsed():
    # every raw unit, re-keyed without the file, minus what parsed already has
    units = [
        banzai.Unit(symbol=k.split("symbol=")[1].split("/")[0], day=k.split("day=")[1].split("/")[0])
        for k in raw.keys()
    ]
    return [u for u in units if u not in parsed.have(units)]


@banzai.job(parsed, feed=unparsed)
def parse(unit):
    csv = raw.store.get(raw.path(banzai.Unit(**unit.fields, file="prices.csv")))
    return pl.read_csv(io.BytesIO(csv)).with_columns(pl.col("ts").str.to_datetime())


if __name__ == "__main__":
    banzai.tick([pull, parse])
```

Now the whole parsed dataset reads back as one lazy frame, with the keys as columns:

```python
parsed.scan().filter(pl.col("symbol") == "AAPL").collect()
```

### 3. A source that revises files in place

Some sources rewrite a file and only tell you through a modified time. Put that time on the unit. A revision is then a new path, so "is it current" is the same existence check as "did I get it". Old versions stay.

```python
@banzai.feed
def changed():
    listing = api.list_files()                       # [(name, modified_at), ...]
    units = [banzai.Unit(updated=ts, file=name) for name, ts in listing]
    return [u for u in units if u not in raw.have(units)]
```

```text
raw/reports/updated=20260901T1204/A44.xml
raw/reports/updated=20260902T1031/A44.xml        # the revision; the first one is still there
```

## Running on S3

Swap the store. Everything else is the same. `prefix` nests the store inside the bucket. The other kwargs go to obstore for bytes and to polars as `storage_options` for frames.

```python
raw = banzai.Dataset("prices", stores.S3Store("my-bucket", prefix="raw", region="us-east-1"))
```

## Running in parallel

`tick` takes a runner. A runner is `runner(banzai_job, units) -> [path | Exception]`, one result per unit. A unit that fails doesn't stop the others: its exception comes back in its slot, tick logs it, and the tick carries on.

```python
from banzai import runners

banzai.tick([pull], runner=runners.parallel)    # joblib, all local cores
```

`tick` returns a `core.Outcome` with `written` paths and `failed` as `(job name, unit, exception)`. A failed unit is just a missing path, so the next tick's feed picks it up again.

For Modal, write a runner in your own code. banzai never imports Modal. `try_run` turns a job failure into a return value, and `return_exceptions=True` does the same for anything Modal itself raises, so both land in the unit's slot.

```python
@app.function(image=image)
def run_unit(banzai_job: banzai.Job, unit: banzai.Unit) -> str | Exception:
    return banzai_job.try_run(unit)

def on_modal(banzai_job, units):
    return list(run_unit.map([banzai_job] * len(units), units, return_exceptions=True))

banzai.tick([pull], runner=on_modal)
```
