SQLite ledger for Sume bulk queues: SKU to run id and what to retry

A Sume bulk queue answers by position: item 0 is the first row you sent. The queue holds no SKU and the API cannot list your queues, so the mapping from index to SKU exists only on your side. A table with a primary key of queue id plus index, a SKU column and the item fields is enough to answer the questions that matter in a campaign: which SKUs have a video, which run_id holds it, and which SKUs still need a retry.

The script below uses only sqlite3 from Python's standard library. It records a queue receipt, then selects the SKUs that failed or were canceled and have never completed in any queue.

The ledger

Call record(queue, skus) with the data object from a poll and the list of SKUs in the order you sent them. The demo at the bottom feeds it a three-item receipt.

import json, sqlite3

db = sqlite3.connect("ledger.db")
db.execute("""CREATE TABLE IF NOT EXISTS item (queue_id TEXT, idx INTEGER, sku TEXT,
              status TEXT, run_id TEXT, error TEXT, PRIMARY KEY (queue_id, idx))""")

def record(queue, skus):
    """Upsert every item of a queue receipt; skus[i] is the SKU sent as item i."""
    with db:
        for it in queue["items"]:
            db.execute("INSERT OR REPLACE INTO item VALUES (?,?,?,?,?,?)",
                       (queue["id"], it["index"], skus[it["index"]], it["status"],
                        it.get("run_id"), it.get("error") and json.dumps(it["error"])))

def to_retry():
    return [r[0] for r in db.execute(
        """SELECT DISTINCT sku FROM item WHERE status IN ('failed','canceled') AND sku NOT IN
           (SELECT sku FROM item WHERE status = 'completed') ORDER BY sku""")]

if __name__ == "__main__":
    q = {"id": "q1", "items": [
        {"index": 0, "status": "completed", "run_id": "r0", "error": None},
        {"index": 1, "status": "failed", "run_id": None, "error": {"code": "format_run_failed"}},
        {"index": 2, "status": "canceled", "run_id": "r2", "error": {"code": "format_run_canceled"}}]}
    record(q, ["A-1", "B-2", "C-3"])
    print(to_retry())

Why the key is queue plus index

Item index restarts at 0 in every queue, so it is only unique inside one. Using INSERT OR REPLACE on that pair makes a re-poll safe: the second write for an item replaces the first one as it moves from queued to running to a terminal state, and writing the same terminal receipt twice changes nothing.

to_retry() excludes any SKU that has a completed row in any queue. That matters after a retry: SKU B-2 fails in queue one, you send it again in queue two, and it completes. The failed row from queue one is still in the table as history, and the query correctly stops listing it.

The error column stores the item's {code, message} as JSON text. For a child that ran, the code is format_run_failed or format_run_canceled, and the reason is on the run receipt, so read GET /v1/format-runs/{run_id} before you retry blindly. For a child that never started, run_id is null and the code names the create failure.

Using it after a partial failure

When the queue reaches completed, call record one last time, then build the next items array from to_retry() and send it under a new idempotency key, since the body is different. Failed items free their slot straight away, so a queue is never held up by a bad row.

Write the ledger row for the queue id before you start polling, with each SKU and a queued status. If your process dies, the queue id is the only handle you have to the batch.

Query the same table for the spend question as well. Join the run_id column to each run receipt, and read usage.debited_usd_micros, which is the real total, rather than billable_amount_usd_micros. That gives a cost per SKU that matches the wallet.