Python 88.3%
TypeScript 7.6%
Shell 4.1%
1#!/usr/bin/env python32"""Batch API end-to-end: build JSONL -> upload (purpose=batch) -> create -> poll -> download output -> map by custom_id -> cleanup.3STATUS: LIVE_VERIFIED 2026-09-18 (validating -> in_progress -> finalizing -> completed in ~30s; cost ≈ $0.000004).4Run: python3 examples/openai/batch/batch-lifecycle.py (set BATCH_ENDPOINT=/v1/chat/completions to try the chat shape)5"""6from __future__ import annotations7import json, os, sys, time, uuid8from pathlib import Path9sys.path.insert(0, str(Path(__file__).resolve().parents[3]))10from scripts.live import openai_request # noqa: E4021112NOTE = "example batch-lifecycle"13ENDPOINT = os.environ.get("BATCH_ENDPOINT", "/v1/responses")14LINES = {15 "/v1/responses": [{"custom_id": "atlas-platform-agent-1", "method": "POST", "url": "/v1/responses",16 "body": {"model": "gpt-5.4-nano", "input": "Reply with OK.", "max_output_tokens": 16}}],17 "/v1/chat/completions": [{"custom_id": "atlas-platform-agent-1", "method": "POST", "url": "/v1/chat/completions",18 "body": {"model": "gpt-4.1-nano", "messages": [{"role": "user", "content": "Reply with OK."}], "max_tokens": 8}}],19 "/v1/embeddings": [{"custom_id": "atlas-platform-agent-1", "method": "POST", "url": "/v1/embeddings",20 "body": {"model": "text-embedding-3-small", "input": "hello world"}}],21}[ENDPOINT]22jsonl = "".join(json.dumps(l) + "\n" for l in LINES).encode()23b = f"----atlas{uuid.uuid4().hex}"24body = (f"--{b}\r\nContent-Disposition: form-data; name=\"purpose\"\r\n\r\nbatch\r\n--{b}\r\n"25 f"Content-Disposition: form-data; name=\"file\"; filename=\"input.jsonl\"\r\nContent-Type: application/jsonl\r\n\r\n").encode() + jsonl + f"\r\n--{b}--\r\n".encode()26st, f, _ = openai_request("POST", "/v1/files", data=body, content_type=f"multipart/form-data; boundary={b}", note=NOTE)27assert st == 200, f28created = [f["id"]]29st, batch, _ = openai_request("POST", "/v1/batches", {"input_file_id": f["id"], "endpoint": ENDPOINT, "completion_window": "24h",30 "metadata": {"owner": "atlas-platform-agent"}}, note=NOTE, est_cost_usd=0.00001)31assert st == 200, batch32print("batch", batch["id"], batch["status"], "expires_at", batch["expires_at"])33t0 = time.time()34while time.time() - t0 < 180 and batch["status"] not in ("completed", "failed", "expired", "cancelled"):35 time.sleep(20)36 st, batch, _ = openai_request("GET", f"/v1/batches/{batch['id']}", note=NOTE + " poll")37 print(f"t+{time.time() - t0:5.1f}s {batch['status']} {batch['request_counts']}")38if batch["status"] not in ("completed", "failed", "expired", "cancelled"):39 st, batch, _ = openai_request("POST", f"/v1/batches/{batch['id']}/cancel", note=NOTE + " cancel")40 print("cancel ->", st, batch.get("status"))41print("usage", batch.get("usage"), "errors", batch.get("errors"))42for key in ("output_file_id", "error_file_id"):43 fid = batch.get(key)44 if fid:45 created.append(fid)46 st, content, _ = openai_request("GET", f"/v1/files/{fid}/content", content_type=None, note=NOTE + f" {key}")47 for line in content.decode().splitlines():48 row = json.loads(line)49 resp = row["response"]50 print(key, row["custom_id"], "->", resp and resp["status_code"], (row.get("error") or {}).get("code"),51 json.dumps((resp or {}).get("body", {}))[:120])52for fid in created:53 st, d, _ = openai_request("DELETE", f"/v1/files/{fid}", note=NOTE + " cleanup")54 print("deleted", fid, d.get("deleted"))55