API Reference

class openai_batch_helper.core.BatchHelper

Bases: object

Helper to manage the OpenAI Batch API.

Example

>>> from openai_batch_helper import BatchHelper
>>> helper = BatchHelper(endpoint="/v1/chat/completions", completion_window="24h")
>>> job = helper.init_job()
>>> _ = job.add_line({
...     "custom_id": "t1",
...     "method": "POST",
...     "url": "/v1/chat/completions",
...     "body": {"model":"gpt-4o-mini","messages":[{"role":"user","content":"hi"}]},
... })
>>> # Submit only when ready:
>>> # job.submit_file().submit_batch_job().wait_for_completion()

__init__(client=None, *, endpoint='/v1/chat/completions', completion_window='24h', workdir=None)

Parameters:

• client (Any | None)

• endpoint (str)

• completion_window (str)

• workdir (str | None)

Return type:

None

init_job(*, filename=None)

Parameters:

filename (str | None)

Return type:

BatchJob

resume_job(batch_id)

Resume an existing batch without resubmitting files.

Parameters:

batch_id (str) – The ID of the batch to resume.

Return type:

BatchJob

class openai_batch_helper.core.BatchJob

Bases: object

Represents a single batch job lifecycle and artifacts.

Methods are chainable to allow a fluent style.

__init__(*, client, endpoint, completion_window, workdir, filename, existing_batch_id=None)

Parameters:

• client (Any)

• endpoint (str)

• completion_window (str)

• workdir (str | None)

• filename (str | None)

• existing_batch_id (str | None)

Return type:

None

add_line(obj_or_json)

Append a single request line to the JSONL file.

Parameters:

obj_or_json (str | Dict[str, Any]) – A Python dict to be JSON-encoded, or a pre-serialized JSON string.

Return type:

BatchJob

add_lines(items)

Parameters:

items (Iterable[str | Dict[str, Any]])

Return type:

BatchJob

add_task(custom_id, url=None, *, body, method='POST')

Append a single request line using convenience parameters.

The url defaults to the job’s endpoint. body is keyword-only to keep argument order unambiguous.

Example

>>> job.add_task("t1", body={
...     "model": "gpt-4o-mini",
...     "messages": [{"role": "user", "content": "hi"}],
... })
>>> job.add_task("emb-1", "/v1/embeddings", body={
...     "model": "text-embedding-3-small",
...     "input": "hello",
... })

Parameters:

• custom_id (str)

• url (str | None)

• body (Dict[str, Any])

• method (str)

Return type:

BatchJob

property batch_id: str | None

cancel()

Return type:

Any

download_errors(dst_path=None)

Parameters:

dst_path (str | None)

Return type:

str | None

download_result(dst_path=None)

Parameters:

dst_path (str | None)

Return type:

str

property error_file_id: str | None

classmethod from_existing(*, client, endpoint, completion_window, workdir, filename, batch_obj)

Parameters:

• client (Any)

• endpoint (str)

• completion_window (str)

• workdir (str | None)

• filename (str | None)

• batch_obj (Any)

Return type:

BatchJob

property input_file_id: str | None

iter_results(results_path=None)

Parameters:

results_path (str | None)

Return type:

Iterator[Dict[str, Any]]

map_by_custom_id(extractor=None, results_path=None)

Return a map of custom_id -> extracted_value.

Default extractor:

• If chat: return response.choices[0].message.content when present.

• If embeddings: return response.data[0].embedding when present.

• Otherwise: return response or { “error”: … }.

Parameters:

• extractor (Callable[[Dict[str, Any]], Any] | None)

• results_path (str | None)

Return type:

Dict[str, Any]

property output_file_id: str | None

property status: str | None

submit_batch_job(*, metadata=None)

Parameters:

metadata (Dict[str, str] | None)

Return type:

BatchJob

submit_file()

Return type:

BatchJob

wait_for_completion(*, poll_seconds=5.0, on_update=None)

Parameters:

• poll_seconds (float)

• on_update (Callable[[Any], None] | None)

Return type:

BatchJob

openai_batch_helper.core.status_progress_logger(logger=None, *, level=20, heartbeat_seconds=30.0)

Return an on_update callback that logs progress via logging.

• Logs immediately on first update (“job submitted”).

• Logs on each status transition.

• Emits heartbeat every heartbeat_seconds even if unchanged (None to disable).

Example

>>> import logging
>>> logging.basicConfig(level=logging.INFO)
>>> job.wait_for_completion(on_update=status_progress_logger())

Parameters:

• logger (Any | None)

• level (int)

• heartbeat_seconds (float | None)

Return type:

Callable[[Any], None]

openai_batch_helper.core.status_progress_printer(stream=None, *, heartbeat_seconds=30.0)

Return an on_update callback that prints progress.

Behavior: - Prints on status transitions immediately. - Additionally, prints a heartbeat line every heartbeat_seconds even if

the status hasn’t changed (set to None to disable heartbeat).

Example

>>> job.wait_for_completion(on_update=status_progress_printer())
>>> # or, more frequent updates
>>> job.wait_for_completion(on_update=status_progress_printer(heartbeat_seconds=10))

Parameters:

• stream (Any | None)

• heartbeat_seconds (float | None)

Return type:

Callable[[Any], None]