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

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.

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:

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.

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:

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.

@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)]

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.

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.

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.

@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)