0Pricing
FastAPI Backend Development Bootcamp · Lektion

Asynchrone Bild- und Dokumenttransformation

Verarbeiten Sie Vorschaubilder, Größenänderungen und Formatkonvertierungen in Background-Workern, um die Request-Latenz niedrig zu halten.

Asynchrone Bild- und Dokumenttransformation ist eine kostenlose FastAPI Backend Development Bootcamp-Lektion auf CoddyKit. Dies ist Lektion 4 von 4. Du kannst die komplette Lektion unten kostenlos lesen – dann übst du sie direkt im Browser mit einem integrierten Code-Editor und einem KI-Tutor rund um die Uhr. Sie ist Teil des FastAPI Backend Development Bootcamp-Lernpfads, und dein Fortschritt wird über Web und CoddyKit-App synchronisiert. Der FastAPI Backend Development Bootcamp-Kurs umfasst insgesamt 4 Lektionen.

Teile dieser Lektion wurden noch nicht übersetzt und werden auf Englisch angezeigt.

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

Häufig gestellte Fragen

Ist die Lektion „Asynchrone Bild- und Dokumenttransformation“ kostenlos?

Ja — der vollständige Text von „Asynchrone Bild- und Dokumenttransformation“ ist hier im Web kostenlos zu lesen. Um sie interaktiv zu üben (integrierter Code-Editor und 24/7 KI-Tutor) und den Rest des FastAPI Backend Development Bootcamp-Kurses freizuschalten, upgrade auf CoddyKit PRO. Der FastAPI Backend Development Bootcamp-Kurs umfasst insgesamt 4 Lektionen.

Was lerne ich in „Asynchrone Bild- und Dokumenttransformation“?

Verarbeiten Sie Vorschaubilder, Größenänderungen und Formatkonvertierungen in Background-Workern, um die Request-Latenz niedrig zu halten. Du übst FastAPI Backend Development Bootcamp mit praktischem Code, den du direkt im Browser ausführst, und ein 24/7 KI-Tutor beantwortet deine Fragen während du die Lektion bearbeitest.

Brauche ich Erfahrung, um FastAPI Backend Development Bootcamp zu starten?

Keine Vorkenntnisse erforderlich. FastAPI Backend Development Bootcamp auf CoddyKit ist für Anfänger bis fortgeschrittene Lernende strukturiert, sodass du hier starten oder von Anfang an beginnen und in deinem eigenen Tempo voranschreiten kannst. Dies ist Lektion 4 von 4.

Wie lange dauert die Lektion „Asynchrone Bild- und Dokumenttransformation“?

Die meisten CoddyKit-Lektionen dauern etwa 5–10 Minuten. Jede ist kompakt und interaktiv, sodass du stetig Fortschritte machst und genau dort weitermachst, wo du aufgehört hast – im Web und in der App.

Kann ich in dieser FastAPI Backend Development Bootcamp-Lektion Code schreiben und ausführen?

Ja. Jede FastAPI Backend Development Bootcamp-Lektion enthält einen integrierten Code-Editor, sodass du echten Code direkt in deinem Browser schreibst und ausführst und sofort KI-Feedback erhältst — ohne lokale Einrichtung erforderlich.

Alle Lektionen in diesem Kurs

  1. Multipart-Uploads und Inhaltsvalidierung
  2. Streaming-Antworten und Range-Requests
  3. Speicher in S3-kompatible Buckets auslagern
  4. Asynchrone Bild- und Dokumenttransformation
← Zurück zu FastAPI Backend Development Bootcamp