Parallel Image Conversion: asyncio + httpx + Idempotency Keys

Sequential image conversion is the most common performance mistake in file processing pipelines. A single requests.post loop processes one image at a time — at 800ms per conversion, a batch of 500 images takes 6.7 minutes. The same batch with 20 concurrent asyncio workers takes under 30 seconds. This guide shows the exact pattern: asyncio + httpx + idempotency keys, with working code and numbers you can quote to stakeholders.

TL;DR — throughput claim

Observed throughput converting PNG → WebP on a standard cloud VM:

Beyond 20 workers you start hitting rate-limit responses (429). Use the retry pattern below to handle them gracefully. Images under 2MB: use these numbers. Larger files: expect 2–4x longer per request.

The serial pattern and its ceiling

Here's what most scripts start with:

import requests
from pathlib import Path

def convert_all(images: list[Path], target: str, api_key: str):
    for img in images:
        resp = requests.post(
            'https://changethisfile.com/v1/convert',
            headers={'Authorization': f'Bearer {api_key}'},
            files={'file': img.open('rb')},
            data={'target': target},
            timeout=60,
        )
        resp.raise_for_status()
        out = img.with_suffix(f'.{target}')
        out.write_bytes(resp.content)
        print(f'Done: {img.name}')

CPU is idle 95% of the time waiting for the network. The fix is concurrency — not threads (GIL overhead, complex error handling) but asyncio with httpx, which gives you clean structured concurrency and connection pooling.

asyncio + httpx: the right concurrency model

Three components make this work:

1. httpx.AsyncClient: Reuses HTTP connections across requests (connection pooling). Without this, each request pays ~100ms for TLS handshake.
2. asyncio.Semaphore: Caps outstanding requests. Without a semaphore, asyncio.gather fires all requests simultaneously — great for 20 files, catastrophic for 10K (exhausts file descriptors, floods the API).
3. Idempotency-Key: Makes every request safe to retry. If the process crashes after the API call but before writing the output file, re-running the job returns the cached result instead of billing a second conversion.

Production code: asyncio + httpx + idempotency

#!/usr/bin/env python3
import asyncio
import hashlib
import os
from pathlib import Path

import httpx

API_KEY = os.environ['CTF_API_KEY']
API_URL = 'https://changethisfile.com/v1/convert'
CONCURRENCY = int(os.environ.get('CONCURRENCY', '10'))


def idempotency_key(file_path: Path, target: str) -> str:
    """Deterministic key: same file + target always maps to same key."""
    payload = f"{file_path.resolve()}|{target}|{file_path.stat().st_mtime_ns}"
    return hashlib.sha256(payload.encode()).hexdigest()[:32]


async def convert_image(
    client: httpx.AsyncClient,
    path: Path,
    target: str,
    sem: asyncio.Semaphore,
) -> tuple[Path, bool, str]:
    """Returns (path, success, error_message)."""
    out_path = path.with_suffix(f'.{target}')
    if out_path.exists():
        return out_path, True, 'skipped (already exists)'

    idem_key = idempotency_key(path, target)

    async with sem:
        for attempt in range(4):
            try:
                async with path.open('rb') as f:
                    content = await f.read()

                resp = await client.post(
                    API_URL,
                    headers={
                        'Authorization': f'Bearer {API_KEY}',
                        'Idempotency-Key': idem_key,
                    },
                    content=content,
                    params={'target': target},
                    timeout=120,
                )

                if resp.status_code == 429:
                    retry_after = int(resp.headers.get('Retry-After', '30'))
                    jitter = attempt * 5
                    await asyncio.sleep(retry_after + jitter)
                    continue

                if resp.status_code >= 500:
                    await asyncio.sleep(2 ** attempt)
                    continue

                resp.raise_for_status()
                out_path.write_bytes(resp.content)
                return out_path, True, ''

            except httpx.TimeoutException:
                await asyncio.sleep(2 ** attempt)
                continue
            except httpx.HTTPStatusError as e:
                return out_path, False, f'HTTP {e.response.status_code}'

    return out_path, False, 'max retries exceeded'


async def batch_convert(paths: list[Path], target: str) -> dict:
    sem = asyncio.Semaphore(CONCURRENCY)
    results = {'success': 0, 'skipped': 0, 'failed': [], 'outputs': []}

    async with httpx.AsyncClient(
        limits=httpx.Limits(max_connections=CONCURRENCY + 2),
        headers={'User-Agent': 'ctf-pipeline/1.0'},
    ) as client:
        tasks = [convert_image(client, p, target, sem) for p in paths]
        for coro in asyncio.as_completed(tasks):
            out_path, ok, msg = await coro
            if ok and msg.startswith('skipped'):
                results['skipped'] += 1
            elif ok:
                results['success'] += 1
                results['outputs'].append(out_path)
            else:
                results['failed'].append({'path': str(out_path), 'error': msg})
                print(f'FAILED: {out_path.name} — {msg}')

    return results


if __name__ == '__main__':
    import sys
    import glob

    if len(sys.argv) < 3:
        print('Usage: python convert.py  ')
        print('Example: python convert.py ./photos/*.heic jpg')
        sys.exit(1)

    files = [Path(p) for p in glob.glob(sys.argv[1])]
    target_fmt = sys.argv[2]

    print(f'Converting {len(files)} files to {target_fmt} ({CONCURRENCY} workers)')
    results = asyncio.run(batch_convert(files, target_fmt))
    print(f"Success: {results['success']}, Skipped: {results['skipped']}, Failed: {len(results['failed'])}")

The same pattern in JavaScript using Node.js fetch:

import { createReadStream } from 'fs';
import { readdir, writeFile } from 'fs/promises';
import { join, extname, basename } from 'path';
import crypto from 'crypto';

const API_KEY = process.env.CTF_API_KEY;
const API_URL = 'https://changethisfile.com/v1/convert';
const CONCURRENCY = parseInt(process.env.CONCURRENCY || '10');

function idempotencyKey(filePath, target) {
  return crypto
    .createHash('sha256')
    .update(`${filePath}|${target}`)
    .digest('hex')
    .slice(0, 32);
}

async function convertImage(filePath, target, semaphore) {
  await semaphore.acquire();
  try {
    const form = new FormData();
    form.append('file', new Blob([createReadStream(filePath)]));
    form.append('target', target);

    const resp = await fetch(API_URL, {
      method: 'POST',
      headers: {
        Authorization: `Bearer ${API_KEY}`,
        'Idempotency-Key': idempotencyKey(filePath, target),
      },
      body: form,
    });

    if (!resp.ok) throw new Error(`HTTP ${resp.status}`);
    const outPath = filePath.replace(extname(filePath), `.${target}`);
    await writeFile(outPath, Buffer.from(await resp.arrayBuffer()));
    return { success: true, outPath };
  } finally {
    semaphore.release();
  }
}

Connection pooling matters more than you think

Without connection reuse, each httpx request does a full TLS handshake (~100ms on a fast connection). With 500 images, that's 50 seconds of pure overhead. httpx.AsyncClient reuses connections automatically when you create a single client instance and pass it to all coroutines — which is why the code above creates the client in batch_convert and passes it down rather than creating a new client per file.

Set max_connections to CONCURRENCY + 2 (small buffer for the index request). The default httpx limit is 100, which is fine up to 100 concurrent workers.

For curl users, --keepalive-time 60 and multiple --parallel requests in a single curl call achieve similar pooling:

ls *.png | parallel -j 10 curl -s -X POST https://changethisfile.com/v1/convert \
  -H "Authorization: Bearer $CTF_API_KEY" \
  -F "file=@{}" -F "target=jpg" \
  -o "{.}.jpg"

Progress tracking and rate observability

Use asyncio.as_completed (not asyncio.gather) when you want live progress — it yields results as they finish rather than waiting for all tasks:

import time

start = time.monotonic()
completed = 0

for coro in asyncio.as_completed(tasks):
    result = await coro
    completed += 1
    rate = completed / (time.monotonic() - start)
    remaining = (total - completed) / rate if rate > 0 else float('inf')
    print(f'\r{completed}/{total} ({rate:.1f}/s, ~{remaining:.0f}s left)', end='')

Watch the X-CTF-Remaining response header. If it drops below 10% of your plan limit mid-batch, pause and decide whether to switch to a higher plan or throttle the batch.

The asyncio + httpx pattern turns a 6-minute serial job into a 30-second parallel one. The semaphore keeps you within rate limits; the idempotency key makes crashes recoverable. Get a free API key and benchmark your specific file types — throughput varies by format and file size.