A Practical Python Pipeline for Batch Image Generation

I prefer treating image generation as a job-processing problem rather than embedding a provider SDK throughout an application.

Put each request into a queue, select a model at routing time, limit concurrency, retry only temporary failures, and save the resulting asset with a manifest record. With a unified API, compatible models can share the same authentication and request path; changing models then becomes a routing decision instead of a new integration.

This walkthrough builds that workflow in Python. It accepts JSON Lines jobs, routes them by type, handles URL and base64 responses, records estimated usage, and writes successful and failed jobs to a manifest.

The Pipeline

The basic flow is:

jobs.jsonl → bounded worker pool → /v1/images/generations → object storage → manifest.jsonl

The queue and storage remain under application control. In this example, the API base URL is https://api.cometapi.com/v1, using one server-side CometAPI key.

The implementation covers the production concerns I usually want in a first version:

• Durable job IDs
• Model selection and catalog validation
• Bounded concurrency
• Exponential backoff with jitter
• URL and base64 image responses
• Estimated per-job cost
• A manifest containing success and failure records

Requirements and Model Selection

You need Python 3.10 or later, the requests package, a key, and a writable output directory.

pip install requests

Set the key on the server rather than putting it in browser code or a repository:

export COMETAPI_KEY="your-key"

The generation endpoint is POST /images/generations under the base URL https://api.cometapi.com/v1.

Before deploying, check the live model catalog. It returns the current model ID, supported endpoint, features, and pricing metadata without requiring an authorization header.

As of August 20, 2026, two useful routes in that catalog were:

Workload	Model ID	Notes
Product images with controlled settings	gpt-image-2	Returns usage data and base64 image content on the documented OpenAI-compatible route
High-volume advertising and content concepts	doubao-seedream-4-5-251128	Uses the same generation route and is listed with per-request pricing

These models should not be assumed to support identical options. Size, quality, format, reference-image handling, and response behavior can differ. I would check the individual model record before sending optional parameters.

Design the Input Around Stable IDs

JSON Lines keeps the input simple. A queue consumer, database export, or spreadsheet conversion can produce the same format:

{"id":"sku-1001","kind":"product","prompt":"Studio product photo of a ceramic coffee dripper on a warm neutral background"}
{"id":"campaign-204","kind":"ad","prompt":"Editorial summer travel image, vivid natural light, wide composition, no text"}
{"id":"blog-088","kind":"content","prompt":"Minimal illustration of a developer automating a creative workflow, no text"}

The id is used for the output filename and manifest key. In a real queue, I would also use it as the idempotency key and skip IDs already marked successful.

The sample routing is:

• product → gpt-image-2
• ad → doubao-seedream-4-5-251128
• content → doubao-seedream-4-5-251128

A job can override the default with its own model field. The script checks the live catalog at startup so an obsolete model ID fails early instead of producing a series of invalid requests.

Concurrency and Retry Behavior

The default worker count is four. That is an application-level starting point, not a universal account limit. I would monitor latency and 429 responses before increasing MAX_WORKERS.

Only these responses are retried:

• 408
• 429
• 5xx

The retry loop uses exponential backoff and jitter. Authentication failures, invalid models, and unsupported parameters are not retried because the request itself must be fixed first.

Save the Actual Image, Not Just a Provider URL

The documented GPT Image response includes data[0].b64_json. Other compatible models may return data[0].url instead.

The worker supports both forms. It writes the result to a temporary path conceptually by completing the download or decode first, then stores the final file. For a production deployment, I would replace the local output/ directory with S3, R2, GCS, or another object store.

A provider-hosted URL should not be treated as permanent storage unless its retention policy explicitly guarantees that.

Complete Python Example

Save this as batch_image_pipeline.py. Put jobs.jsonl beside it, then run python3 batch_image_pipeline.py.

import base64, json, os, random, time
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
import requests

BASE_URL = "https://api.cometapi.com/v1"
KEY = os.environ["COMETAPI_KEY"]
WORKERS = int(os.getenv("MAX_WORKERS", "4"))
OUT = Path("output")
ROUTES = {
    "product": "gpt-image-2",
    "ad": "doubao-seedream-4-5-251128",
    "content": "doubao-seedream-4-5-251128",
}

catalog = requests.get("https://api.cometapi.com/api/models", timeout=30)
catalog.raise_for_status()
CATALOG = {model["id"]: model for model in catalog.json()["data"]}

def generate(job):
    model = job.get("model", ROUTES[job["kind"]])
    if model not in CATALOG:
        raise ValueError(f"Unknown model: {model}")
    payload = {"model": model, "prompt": job["prompt"], "n": 1}
    if model == "gpt-image-2":
        payload.update(quality="low", size="1024x1024", output_format="jpeg")

    for attempt in range(4):
        response = requests.post(
            f"{BASE_URL}/images/generations",
            headers={"Authorization": f"Bearer {KEY}"},
            json=payload,
            timeout=180,
        )
        if response.status_code not in {408, 429} and response.status_code < 500:
            break
        time.sleep(2**attempt + random.random())
    response.raise_for_status()
    body = response.json()
    item = body["data"][0]

    if item.get("b64_json"):
        data = base64.b64decode(item["b64_json"])
        extension = body.get("output_format", "png")
    else:
        download = requests.get(item["url"], timeout=120)
        download.raise_for_status()
        data = download.content
        extension = {"image/png": "png", "image/webp": "webp"}.get(
            download.headers.get("content-type"), "jpg"
        )
    path = OUT / f"{job['id']}.{extension}"
    path.write_bytes(data)

    price, usage = CATALOG[model].get("pricing") or {}, body.get("usage", {})
    cost = price.get("per_request")
    if cost is None and price.get("input") is not None:
        cost = (usage.get("input_tokens", 0) * price["input"] +
                usage.get("output_tokens", 0) * price["output"]) / 1_000_000
    return {"id": job["id"], "model": model, "path": str(path),
            "estimated_usd": cost * price.get("ratio", 1) if cost is not None else None}

def safe_generate(job):
    try:
        return {"status": "success", **generate(job)}
    except Exception as error:
        return {"id": job["id"], "status": "failed", "error": str(error)}

OUT.mkdir(exist_ok=True)
jobs = [json.loads(line) for line in Path("jobs.jsonl").read_text().splitlines() if line]
with ThreadPoolExecutor(max_workers=WORKERS) as pool:
    results = list(pool.map(safe_generate, jobs))
with (OUT / "manifest.jsonl").open("w") as manifest:
    manifest.writelines(json.dumps(result) + "\n" for result in results)

The catalog is fetched at runtime, but the fallback routes in ROUTES were verified on August 20, 2026. They should be checked again before deploying on another date.

Run a Small Smoke Test First

Start with one worker and one job:

MAX_WORKERS=1 python3 batch_image_pipeline.py

A successful GPT Image response has this shape:

{
  "created": 1776841943,
  "output_format": "jpeg",
  "quality": "low",
  "size": "1024x1024",
  "usage": {
    "input_tokens": 16,
    "output_tokens": 208,
    "total_tokens": 224
  },
  "data": [{"b64_json": ""}]
}

The script decodes this response and writes output/.jpeg. It then adds a success row to output/manifest.jsonl. URL-based responses are downloaded and represented in the same manifest format.

The code was syntax-checked locally, but a live generation request still requires a valid key. I would run this one-job test before increasing concurrency.

Estimating Cost

Pricing is time-sensitive. On August 20, 2026, the live catalog returned these base values and a 0.8 billing ratio:

• gpt-image-2: $5 per 1M input tokens and $30 per 1M output tokens
• Effective rates after the listed ratio: $4 per 1M input tokens and $24 per 1M output tokens
• doubao-seedream-4-5-251128: $0.04 per request
• Effective request price after the listed ratio: $0.032

The pricing guide describes token-based billing for models with official token pricing and request-based billing for models priced per call.

The script applies these calculations:

token cost = ratio × (input tokens × input rate + output tokens × output rate) / 1,000,000
request cost = ratio × per-request price

For the documented GPT Image response with 16 input tokens and 208 output tokens, the August 20 catalog values produce an illustrative estimate of about $0.005056.

The actual bill depends on the model, prompt, quality, size, response usage, and retries. Account usage and API responses should be treated as the billing record rather than a fixed per-image estimate.

I also budget for failed or rejected work. A retry after an uncertain timeout may create a second billable result. A technically successful image can still fail review. A useful operational metric is:

effective cost per accepted image = total batch spend / approved images

Troubleshooting Guide

Symptom	Likely cause	Fix
401	Missing or invalid key	Check the server-side COMETAPI_KEY
400	Invalid model or unsupported option	Recheck the live catalog and remove model-specific fields
429	Excessive concurrency	Lower MAX_WORKERS and retain exponential backoff
Repeated 5xx	Temporary upstream failure	Retry with a cap, then use a dead-letter queue
No saved image	Different response container	Inspect data[0] and support b64_json or url
Duplicate spend	Replay after partial failure	Use durable IDs and acknowledge only after storage succeeds

A permanent 400 will not become valid through retries. Likewise, an unlimited 429 loop can turn a traffic spike into a growing backlog.

What I Would Change for Production

For multiple workers, I would replace JSON Lines with a durable queue. Set the visibility timeout longer than the maximum generation time, acknowledge only after both the image and manifest are stored, and send exhausted jobs to a dead-letter queue.

Keep optional parameters model-specific. The shared payload should contain only fields such as model, prompt, and n: 1. Add quality, size, or output_format only when the selected model documentation supports them. If fallback routing is added, rebuild the payload for the fallback model rather than blindly reusing provider-specific options.

Other practical safeguards include:

• Store keys in a secret manager.
• Restrict and validate prompt input.
• Scan generated assets according to your policy.
• Keep provider URLs out of long-term product records.
• Log job ID, model ID, latency, attempts, usage, storage path, review result, and catalog snapshot date.
• Set a maximum batch size and per-job retry limit.
• Add daily spend alerts and approval-rate stop conditions.

The key metric is not headline price. It is accepted-image cost after retries, failures, post-processing, and review.

Further References

For endpoint and response details, use the Quick Start, model catalog documentation, image generation reference, and pricing guide.