0Pricing
FastAPI Backend Development Bootcamp · Lesson

Async Image and Document Transformation

Process thumbnails, resizing, and format conversion in background workers to keep request latency low.

Async Image and Document Transformation is a free FastAPI Backend Development Bootcamp lesson on CoddyKit — lesson 4 of 4. You can read the complete lesson below for free — then practise it hands-on in the browser with a built-in code editor and a 24/7 AI tutor. It is part of the FastAPI Backend Development Bootcamp learning path, one of 4 lessons in the course, and your progress syncs across the web and the CoddyKit app.

Why Offload Media Work

Resizing an image or converting a PDF can take hundreds of milliseconds to several seconds. If you do that work inside the request handler, the client waits and your worker process is blocked.

The pattern for B2-level FastAPI services is:

  • Accept the upload, persist the original quickly
  • Return a 202 Accepted with a job id
  • Do thumbnails, resizing, and format conversion in a background worker

This keeps request latency low and makes heavy CPU work independently scalable.

BackgroundTasks vs a Real Queue

FastAPI ships BackgroundTasks, which runs a function after the response is sent but still inside the same process. It is fine for cheap, fast follow-ups (sending an email, writing a log).

For CPU-heavy media transforms it is the wrong tool: it competes with your event loop and dies if the process restarts. Prefer a dedicated task queue (Celery, RQ, Dramatiq, or arq) backed by Redis so work survives deploys and scales horizontally.

from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

def log_upload(filename: str) -> None:
    # cheap follow-up work only
    print(f"received {filename}")

@app.post("/upload")
async def upload(background: BackgroundTasks):
    background.add_task(log_upload, "photo.png")
    return {"status": "accepted"}

Accept Fast, Process Later

The endpoint should do the minimum: validate the file, stream it to storage, create a job row, and enqueue a task. Notice we read the upload in chunks so a large file never loads fully into memory.

  • await file.read(chunk) avoids huge memory spikes
  • We return a job_id the client can poll
  • The actual transform happens in process_image.delay(...)
import uuid, aiofiles
from fastapi import FastAPI, UploadFile, status

app = FastAPI()

@app.post("/images", status_code=status.HTTP_202_ACCEPTED)
async def create_image(file: UploadFile):
    job_id = str(uuid.uuid4())
    dest = f"/data/originals/{job_id}_{file.filename}"
    async with aiofiles.open(dest, "wb") as out:
        while chunk := await file.read(1024 * 1024):
            await out.write(chunk)
    process_image.delay(job_id, dest)  # enqueue
    return {"job_id": job_id, "status": "queued"}

Generating Thumbnails with Pillow

Pillow's Image.thumbnail() resizes in place while preserving aspect ratio and never upscaling. It is the right primitive for thumbnails because the result fits within the box you give it.

Use Image.LANCZOS resampling for sharp downscales, and call img.convert("RGB") before saving as JPEG so images with alpha channels (PNG) do not crash the encoder.

from PIL import Image

def make_thumbnail(src: str, dst: str, box=(256, 256)) -> None:
    with Image.open(src) as img:
        img = img.convert("RGB")
        img.thumbnail(box, Image.LANCZOS)
        img.save(dst, "JPEG", quality=85, optimize=True)

if __name__ == "__main__":
    print("thumbnail helper ready")

A Celery Worker Task

Each transform becomes a Celery task. The task is a plain function decorated with @app.task; the queue handles retries, acknowledgements, and concurrency.

  • Generate multiple sizes in one task to amortize the image decode
  • Update the job status when done so the API can report progress
  • Set autoretry_for so transient I/O errors retry automatically
from celery import Celery
from PIL import Image

celery_app = Celery("media", broker="redis://localhost:6379/0")

SIZES = {"thumb": (256, 256), "medium": (1024, 1024)}

@celery_app.task(autoretry_for=(OSError,), retry_backoff=True, max_retries=3)
def process_image(job_id: str, src: str) -> dict:
    outputs = {}
    with Image.open(src) as base:
        base = base.convert("RGB")
        for name, box in SIZES.items():
            img = base.copy()
            img.thumbnail(box, Image.LANCZOS)
            dst = f"/data/derived/{job_id}_{name}.jpg"
            img.save(dst, "JPEG", quality=85, optimize=True)
            outputs[name] = dst
    return {"job_id": job_id, "outputs": outputs}

Format Conversion: PNG and WebP

Serving WebP instead of JPEG/PNG cuts payload size 25-35% with similar quality, which lowers bandwidth and speeds page loads.

Pillow converts by simply choosing the output format in save(). Keep an original-format copy too, since some old clients cannot decode WebP. The example below produces both a JPEG and a WebP from one decode.

from PIL import Image

def to_jpeg_and_webp(src: str, stem: str) -> dict:
    with Image.open(src) as img:
        rgb = img.convert("RGB")
        jpeg_path = f"{stem}.jpg"
        webp_path = f"{stem}.webp"
        rgb.save(jpeg_path, "JPEG", quality=85, optimize=True)
        rgb.save(webp_path, "WEBP", quality=80, method=6)
    return {"jpeg": jpeg_path, "webp": webp_path}

if __name__ == "__main__":
    print(to_jpeg_and_webp.__name__)

Tracking Job Status

Clients need to know when their derivatives are ready. Store a small status record (in Redis or your DB) keyed by job_id and expose a polling endpoint.

Lifecycle states are typically queued -> processing -> done or failed. The worker updates the record at the start and end of the task; the API just reads it.

import json, redis

r = redis.Redis()

def set_status(job_id: str, state: str, **extra) -> None:
    payload = {"state": state, **extra}
    r.set(f"job:{job_id}", json.dumps(payload), ex=86400)

def get_status(job_id: str) -> dict | None:
    raw = r.get(f"job:{job_id}")
    return json.loads(raw) if raw else None

Polling Endpoint and Result URLs

The status endpoint returns the current state and, once done, the URLs of the generated assets. Return 404 for an unknown job and 200 with the state otherwise.

A common upgrade is to return a pre-signed S3 URL for each derivative so the client downloads directly from object storage instead of through your API.

from fastapi import FastAPI, HTTPException

app = FastAPI()

@app.get("/images/{job_id}")
async def image_status(job_id: str):
    status = get_status(job_id)
    if status is None:
        raise HTTPException(status_code=404, detail="job not found")
    return {"job_id": job_id, **status}

Keeping the Event Loop Unblocked

Even outside a worker, you sometimes must call a blocking library (Pillow, a PDF tool) from an async endpoint. Calling it directly blocks the event loop and stalls every concurrent request.

Offload it to a thread pool with asyncio.to_thread (or Starlette's run_in_threadpool). For CPU-bound batches across cores, a ProcessPoolExecutor avoids the GIL. The runnable example shows the thread-offload pattern.

import asyncio, time

def blocking_resize(n: int) -> int:
    time.sleep(0.1)  # stand-in for Pillow work
    return n * n

async def handle(n: int) -> int:
    # runs blocking_resize in a worker thread, loop stays free
    return await asyncio.to_thread(blocking_resize, n)

async def main() -> None:
    results = await asyncio.gather(*(handle(i) for i in range(5)))
    print(results)

if __name__ == "__main__":
    asyncio.run(main())

Converting Documents to PDF/Images

Document transforms (DOCX to PDF, PDF page to PNG thumbnail) usually shell out to external tools like LibreOffice (soffice --headless) or pdftoppm. These are heavy and slow, so they belong in a worker task, never in the request path.

Always run them with a timeout and capture errors, because external converters can hang on malformed input.

import subprocess

def docx_to_pdf(src: str, out_dir: str) -> str:
    subprocess.run(
        ["soffice", "--headless", "--convert-to", "pdf",
         "--outdir", out_dir, src],
        check=True, timeout=120,
    )
    return out_dir

@celery_app.task(autoretry_for=(subprocess.TimeoutExpired,), max_retries=2)
def convert_document(job_id: str, src: str) -> dict:
    out = docx_to_pdf(src, "/data/derived")
    set_status(job_id, "done", out_dir=out)
    return {"job_id": job_id, "out_dir": out}

Validation, Limits, and Cleanup

Untrusted media is a security surface. Protect the pipeline before any heavy work runs:

  • Verify type by content (e.g. Image.open().verify() or magic bytes), not just the file extension
  • Cap dimensions to defuse decompression-bomb images; set Image.MAX_IMAGE_PIXELS
  • Enforce size limits while streaming the upload
  • Clean up originals and derivatives on failure or after a TTL

Reject bad input early so a malicious file never reaches the worker.

from PIL import Image, UnidentifiedImageError

Image.MAX_IMAGE_PIXELS = 50_000_000  # guard against decompression bombs

def is_safe_image(path: str) -> bool:
    try:
        with Image.open(path) as img:
            img.verify()  # checks integrity without full decode
        return True
    except (UnidentifiedImageError, OSError):
        return False

Quick Check

Test your understanding of where heavy media work belongs.

Recap

You learned how to keep media-heavy FastAPI endpoints fast:

  • Accept fast, process later: stream the upload to storage, return 202 with a job id, enqueue the work
  • Use a real queue (Celery/RQ/arq + Redis) for CPU-heavy transforms; reserve BackgroundTasks for cheap follow-ups
  • Transform with Pillow: thumbnail() for aspect-preserving resizes, convert("RGB") before JPEG, and WebP for smaller payloads
  • Document conversion shells out to tools like LibreOffice with a timeout, always in a worker
  • Never block the loop: offload stray blocking calls with asyncio.to_thread
  • Guard input: verify type, cap pixels, limit size, and clean up derivatives

Frequently asked questions

Is the “Async Image and Document Transformation” lesson free?

Yes — the full text of “Async Image and Document Transformation” is free to read here on the web, and the FastAPI Backend Development Bootcamp course includes 4 lessons in total. To practise it interactively (a built-in code editor and a 24/7 AI tutor) and unlock the rest of the FastAPI Backend Development Bootcamp course, upgrade to CoddyKit PRO.

What will I learn in “Async Image and Document Transformation”?

Process thumbnails, resizing, and format conversion in background workers to keep request latency low. You practise FastAPI Backend Development Bootcamp with hands-on code you run directly in the browser, and a 24/7 AI tutor answers your questions as you work through the lesson.

Do I need any experience to start FastAPI Backend Development Bootcamp?

No prior experience is required. FastAPI Backend Development Bootcamp on CoddyKit is structured for beginners through advanced learners; this is — lesson 4 of 4, so you can start here or from the beginning and move at your own pace.

How long does the “Async Image and Document Transformation” lesson take?

Most CoddyKit lessons take about 5–10 minutes. Each one is bite-sized and interactive, so you make steady progress and pick up exactly where you left off across the web and the app.

Can I write and run code in this FastAPI Backend Development Bootcamp lesson?

Yes. Every FastAPI Backend Development Bootcamp lesson includes a built-in code editor, so you write and run real code right in your browser and get instant AI feedback — no local setup required.

All lessons in this course

  1. Multipart Uploads and Content Validation
  2. Streaming Responses and Range Requests
  3. Offloading Storage to S3-Compatible Buckets
  4. Async Image and Document Transformation
← Back to FastAPI Backend Development Bootcamp