0Pricing
FastAPI Backend Development Bootcamp · Leçon

Transformation asynchrone d’images et de documents

Traitez les vignettes, le redimensionnement et la conversion de format dans des travailleurs en arrière-plan afin de maintenir une faible latence des requêtes.

Transformation asynchrone d’images et de documents est une leçon FastAPI Backend Development Bootcamp gratuite sur CoddyKit. Ceci est la leçon 4 sur 4. Tu peux lire la leçon complète ci-dessous gratuitement — puis la pratiquer en direct dans le navigateur avec un éditeur de code intégré et un tuteur IA 24/7. Elle fait partie du parcours d'apprentissage FastAPI Backend Development Bootcamp, et ta progression se synchronise sur le web et l'application CoddyKit. Le cours FastAPI Backend Development Bootcamp comprend 4 leçons au total.

Certaines parties de cette leçon n'ont pas encore été traduites et s'affichent en anglais.

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

Questions Fréquemment Posées

La leçon « Transformation asynchrone d’images et de documents » est-elle gratuite ?

Oui — le texte complet de « Transformation asynchrone d’images et de documents » est gratuit à lire ici sur le web. Pour la pratiquer de manière interactive (un éditeur de code intégré et un tuteur IA 24/7) et déverrouiller le reste du cours FastAPI Backend Development Bootcamp, passe à CoddyKit PRO. Le cours FastAPI Backend Development Bootcamp comprend 4 leçons au total.

Qu'est-ce que j'apprendrai dans « Transformation asynchrone d’images et de documents » ?

Traitez les vignettes, le redimensionnement et la conversion de format dans des travailleurs en arrière-plan afin de maintenir une faible latence des requêtes. Tu pratiques FastAPI Backend Development Bootcamp avec du code pratique que tu exécutes directement dans le navigateur, et un tuteur IA 24/7 répond à tes questions au fur et à mesure que tu avances dans la leçon.

Dois-je avoir de l'expérience pour commencer FastAPI Backend Development Bootcamp ?

Aucune expérience préalable n'est requise. FastAPI Backend Development Bootcamp sur CoddyKit est structuré pour les débutants jusqu'aux apprenants avancés, donc tu peux commencer ici ou depuis le début et avancer à ton rythme. Ceci est la leçon 4 sur 4.

Combien de temps prend la leçon « Transformation asynchrone d’images et de documents » ?

La plupart des leçons CoddyKit prennent environ 5–10 minutes. Chacune est courte et interactive, tu progresses régulièrement et tu repiques exactement où tu t'es arrêté sur le web et l'app.

Peux-tu écrire et exécuter du code dans cette leçon FastAPI Backend Development Bootcamp ?

Oui. Chaque leçon FastAPI Backend Development Bootcamp inclut un éditeur de code intégré, tu écris et exécutes du vrai code directement dans ton navigateur et tu reçois des retours IA instantanés — aucune configuration locale requise.

Toutes les leçons de ce cours

  1. Téléversements multiparties et validation du contenu
  2. Réponses en flux continu et requêtes par plages
  3. Délégation du stockage vers des compartiments compatibles S3
  4. Transformation asynchrone d’images et de documents
← Retour à FastAPI Backend Development Bootcamp