0Pricing
FastAPI Backend Development Bootcamp · Aula

Transformação assíncrona de imagens e documentos

Processe miniaturas, redimensionamentos e conversões de formato em workers em segundo plano para manter baixa a latência das requisições.

Transformação assíncrona de imagens e documentos é uma aula grátis de FastAPI Backend Development Bootcamp no CoddyKit. Esta é a aula 4 de 4. Você pode ler a aula completa abaixo gratuitamente — depois pratica ao vivo no navegador com um editor de código integrado e um tutor de IA 24/7. Faz parte do caminho de aprendizado de FastAPI Backend Development Bootcamp, e seu progresso é sincronizado entre a web e o app CoddyKit. O curso de FastAPI Backend Development Bootcamp inclui 4 aulas no total.

Partes desta aula ainda não foram traduzidas e aparecem em inglês.

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

Perguntas Frequentes

A aula “Transformação assíncrona de imagens e documentos” é grátis?

Sim — o texto completo de “Transformação assíncrona de imagens e documentos” é grátis para ler aqui na web. Para praticá-la interativamente (um editor de código integrado e um tutor de IA 24/7) e desbloquear o restante do curso de FastAPI Backend Development Bootcamp, atualize para CoddyKit PRO. O curso de FastAPI Backend Development Bootcamp inclui 4 aulas no total.

O que vou aprender em “Transformação assíncrona de imagens e documentos”?

Processe miniaturas, redimensionamentos e conversões de formato em workers em segundo plano para manter baixa a latência das requisições. Você pratica FastAPI Backend Development Bootcamp com código prático que executa diretamente no navegador, e um tutor de IA 24/7 responde suas dúvidas enquanto trabalha na aula.

Preciso ter experiência prévia para começar FastAPI Backend Development Bootcamp?

Nenhuma experiência prévia é necessária. FastAPI Backend Development Bootcamp no CoddyKit é estruturado para alunos iniciantes até avançados, então você pode começar aqui ou desde o início e aprender no seu ritmo. Esta é a aula 4 de 4.

Quanto tempo leva a aula “Transformação assíncrona de imagens e documentos”?

A maioria das aulas CoddyKit leva cerca de 5–10 minutos. Cada uma é compacta e interativa, então você faz progresso constante e retoma exatamente de onde parou entre web e app.

Posso escrever e executar código nesta aula de FastAPI Backend Development Bootcamp?

Sim. Cada aula de FastAPI Backend Development Bootcamp inclui um editor de código integrado, então você escreve e executa código real direto no navegador e recebe feedback de IA instantaneamente — nenhuma configuração local necessária.

Todas as aulas deste curso

  1. Uploads multipartes e validação de conteúdo
  2. Respostas em transmissão e requisições por intervalo
  3. Delegação do armazenamento para buckets compatíveis com S3
  4. Transformação assíncrona de imagens e documentos
← Voltar para FastAPI Backend Development Bootcamp