batchpipe

A small, reusable OpenAI Batch API pipeline: submit → fetch → save (batch jobs run at ~50% of live-API cost). The mechanics — request JSONLs, uploads, job tracking, polling, downloads, result parsing, de-duplication — live in the library. Each project only supplies a Pydantic schema, a row selector and a save function. All three stages are idempotent and safe to re-run.

Install

uv add git+https://github.com/hcss-utils/batchpipe.git
# or pin a tag: uv add "git+https://github.com/hcss-utils/batchpipe.git@v0.1.0"

OPENAI_API_KEY is read from the environment (no keys live in this library):

uv run --env-file .env python my_task.py ...

Usage

Wire a pipeline once, then make_cli gives you the three subcommands:

from pathlib import Path
from pydantic import BaseModel
from batchpipe import BatchPipeline, make_cli, setup_logging


class Classification(BaseModel):
    label: str
    confidence: int


setup_logging(Path("batch.log"))

pipeline = BatchPipeline(
    name="my_task",
    model="gpt-5-mini",
    system_prompt="Classify the text...",
    response_model=Classification,
    data_dir=Path("./.cache/my_task"),
)


def select():
    # (custom_id, content) pairs; custom_id maps a result back to your row.
    return [(f"row-{i}", text) for i, text in load_rows()]


def save_one(custom_id: str, result: Classification):
    write_to_db(int(custom_id.removeprefix("row-")), result)


main = make_cli(pipeline, select=select, save_handler=save_one)

if __name__ == "__main__":
    raise SystemExit(main())

Commands

# 1. build request files, upload, create batch jobs
uv run --env-file .env python my_task.py submit --limit 1000

# 2. poll the jobs and download finished outputs (--watch loops until all settle)
uv run --env-file .env python my_task.py fetch --watch

# 3. parse downloaded results and persist them
uv run --env-file .env python my_task.py save

API (if you skip make_cli)

pipeline.submit(items, batch_size=40_000, limit=None)  # items: (custom_id, content)
pipeline.fetch(watch=False, interval=300)
pipeline.save(handler, should_skip=already_done.__contains__)
for custom_id, result in pipeline.iter_results():       # lower-level
    ...

• response_format(model) — strict json_schema block, if you build requests yourself.
• setup_logging(log_file=None) — opt-in handlers for the batchpipe logger.

To match an existing on-disk layout exactly (e.g. when replacing a hand-written script), override the artifact names:

BatchPipeline(
    name="mw", ...,
    input_prefix="mw_input_", results_prefix="mw_results_",
    error_prefix="mw_errors_", ids_filename="batch_ids.jsonl",
)

Requires Python ≥ 3.10, openai, pydantic.