Bigjpg API Tutorial: Batch Upscale Images with Python

The web interface is fine for a handful of images. For a folder of five hundred, you want the API: four endpoints, no browser, and a script you can re-run. This is the version we would hand to a colleague — upload, create, poll, download, with the failure handling that matters in practice.

Before you start

• API access is a paid-account feature. Your key is on the API reference page once you are logged in with an active plan.
• Python 3 with requests installed. Nothing else.
• Enlargement runs asynchronously on our workers, so the script creates tasks, then polls. A 500-image job takes roughly as long as the slowest queue you are in, not the sum of the images.

The script

import json
import os
import time
import requests

API = "https://bigjpg.com"
KEY = "YOUR_API_KEY"
STYLE = "art"          # "art" for illustration, "photo" for photographs
NOISE = "1"            # -1 none, 0 low, 1 medium, 2 high, 3 highest
X2 = "2"               # 1 = 2x, 2 = 4x, 3 = 8x, 4 = 16x
POLL_SECONDS = 8
MAX_RETRY = 1

HEADERS = {"X-API-KEY": KEY}


def upload(path):
    """Get a presigned URL, PUT the file, return its public URL."""
    name = path.split("/")[-1]
    r = requests.post(
        API + "/api/img_upload",
        headers=HEADERS,
        data={"filename": name, "filetype": "image/jpeg"},
        timeout=60,
    )
    info = r.json()
    if info.get("status") != "ok":
        raise RuntimeError("upload failed: " + str(info))

    with open(path, "rb") as image:
        r = requests.put(
            info["url"],
            data=image,
            headers={"Content-Type": "image/jpeg"},
            timeout=300,
        )
    if r.status_code != 200:
        raise RuntimeError("upload failed with HTTP %s" % r.status_code)
    return info["fileurl"]


def create(image_url):
    """Create an enlargement task, return its task id."""
    payload = {
        "style": STYLE,
        "noise": NOISE,
        "x2": X2,
        "input": image_url,
    }
    r = requests.post(
        API + "/api/task/",
        headers=HEADERS,
        data=json.dumps(payload),
        timeout=60,
    )
    data = r.json()
    if "tid" not in data:
        raise RuntimeError("task failed: " + str(data))
    return data["tid"]


def poll(tids):
    """Query up to 100 task ids at once. Returns task id -> result dict."""
    r = requests.get(API + "/api/task/" + ",".join(tids), timeout=60)
    return r.json()


def download(url, out):
    """Download a finished image, checking the transfer really succeeded."""
    r = requests.get(url, timeout=300)
    if r.status_code != 200:
        raise RuntimeError("download failed with HTTP %s" % r.status_code)
    with open(out, "wb") as f:
        f.write(r.content)
    return out


def main(paths):
    pending = {}
    attempts = {}
    for path in paths:
        try:
            tid = create(upload(path))
        except Exception as e:
            print("skipped", path, e)
            continue
        pending[tid] = path
        attempts[tid] = 0
        print("queued", path, tid)

    while pending:
        for tid, info in poll(list(pending)[:100]).items():
            if info.get("status") == "success":
                out = os.path.splitext(pending[tid])[0] + "_bigjpg.jpg"
                try:
                    download(info["url"], out)
                except Exception as e:
                    print("download failed", pending[tid], e)
                    pending.pop(tid)
                    continue
                print("done", pending.pop(tid), "->", out)

            elif info.get("status") == "error":
                # One retry, then give up on this file.
                attempts[tid] += 1
                if attempts[tid] <= MAX_RETRY:
                    requests.post(API + "/api/task/" + tid, headers=HEADERS, timeout=60)
                    print("retrying", pending[tid])
                else:
                    print("giving up on", pending.pop(tid))

        if pending:
            time.sleep(POLL_SECONDS)


if __name__ == "__main__":
    import sys
    main(sys.argv[1:])

What each call does

Concurrency and manners

• Poll less often than you think. Enlargement is not instant; 5–10 seconds between polls is plenty, and querying 100 ids in one call keeps the request count low.
• Cap your uploads. Two to four uploads in flight is a reasonable ceiling for a single machine. If you need more throughput, run more instances rather than more threads per instance.
• Retry with backoff. A network error is not a task failure. Retry the HTTP call up to three times with a growing delay before you mark a file as broken.
• Watch remaining_api_calls. Every created task reports it; stop the loop at some low threshold instead of discovering the limit from a wall of errors.
• Log the mapping. Keep a file with tid, input name, settings and output name. When a job runs overnight, that log is the only way to know what finished.

Errors you will actually hit

• valid_api_key_required — the header is missing or unknown. Check the header name and that the key belongs to a plan that has not expired.
• requires_vip — free or expired account. The API is for active paid plans.
• month_limit — the monthly quota is used up. Each task reports the remainder, so you can stop before this happens.
• param_error — style must be art or photo, noise -1 to 3, x2 1 to 4.
• input_url_error — the input has to be a publicly reachable http or https image URL. Presigned private URLs expire; upload first, then pass the returned public URL.

What a batch costs

One image counts as one credit at 2x, 4x and 8x; 16x counts as two. So a 500-image job is 500 credits at 4x, and the same job at 16x is 1000. Since 4x is the right factor for most photographs anyway, the cheap option is also the one that looks better — see the factor guide for the reasoning.

The full endpoint reference, including parameters and every error status, is on the API page. If something in this script does not match what you get back, that page is the version we keep current.