BigQuery product table to Sume bulk queues and a queue map

Query the products you want as videos, cast each column to a plain string or integer, and send them to Sume 100 rows at a time. Each chunk becomes one bulk queue (POST /v1/formats/{handle}/{slug}/bulk-runs). Then write queue_id, idx and sku to a small BigQuery table. That table is how a finished run is joined back to a product, because the queue reports items by position.

Python's json.dumps raises TypeError on a Decimal or a date object. Casting in SQL means every value is already a string or an integer before it reaches the HTTP body, so the query is the one place to fix types.

Query and body shape

Order the query by SKU so the chunk boundaries are the same on every run. A different order moves rows between chunks, and a chunk key that was already used with a different body returns 409 idempotency_conflict.

Each item is the same shape as a single Format run. Put row data in input, which is a JSON object of at most 64 top-level keys and 2 MiB, and keep the task text short in instruction. Sume passes input to the agent as data and not as instructions, so scraped or supplier-written text belongs there.

The script

BATCH is part of every key. insert_rows_json returns a list of row errors and does not raise, so the script asserts on it.

import json, os, urllib.request
from google.cloud import bigquery

SQL = ("SELECT sku, product_url, CAST(price AS STRING) AS price "
       "FROM `shop.catalog.bf_2026` WHERE active ORDER BY sku")
URL = f"https://api.sume.com/v1/formats/{os.environ['SUME_FORMAT']}/bulk-runs"
BATCH = "bf-2026-v1"

def create_queue(items, n):
    req = urllib.request.Request(URL, method="POST",
        data=json.dumps({"concurrency": 4, "items": items}).encode(),
        headers={"Authorization": "Bearer " + os.environ["SUME_API_KEY"],
                 "Content-Type": "application/json",
                 "Idempotency-Key": f"{BATCH}-chunk-{n}"})
    with urllib.request.urlopen(req) as r:
        return json.load(r)["data"]["id"]

client = bigquery.Client()
rows = [dict(r) for r in client.query(SQL).result()]
for n in range(0, len(rows), 100):
    chunk = rows[n:n + 100]
    items = [{"instruction": "Make a 9:16 product ad.", "input": r} for r in chunk]
    qid = create_queue(items, n // 100)
    errs = client.insert_rows_json("shop.ops.sume_queue_items",
        [{"queue_id": qid, "idx": i, "sku": r["sku"]} for i, r in enumerate(chunk)])
    assert not errs, errs

Reading results later

Poll each queue at status_url and branch on counts. A queue is completed when every item is terminal, and that still includes failed items. For one failed row, the item error only says format_run_failed; the cause is on the run receipt at GET /v1/format-runs/{run_id}.

Reads and writes have separate rate budgets per key, and reads get forty times the write number. One create per chunk costs one write, so a 250-row table is three writes, and polls come out of the read budget.

If the script stops halfway

Run it again with the same BATCH. The queues that were already created come back as 202 with the original queue, so no second run starts. The map insert then writes the same rows again, so give that table a unique key on (queue_id, idx) or load it with a merge. A lost queue id can be recovered the same way, by replaying the create call with the same key and the same body.

The key and its scope matter here: an idempotency key belongs to one Format, so the same string sent to a second Format starts a second set of runs. If one table feeds two Formats, put the Format slug in the key as well as the batch label and chunk number.