PricingLoginStart free
FastAPI Backend Development Bootcamp · 课时

异步图像与文档转换

在后台工作进程中处理缩略图、大小调整和格式转换，降低请求延迟。

第 4 / 4 课13 个步骤

异步图像与文档转换 是 CoddyKit 上的免费 FastAPI Backend Development Bootcamp 课时。 这是第 4 节课，共 4 节。 你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。 这是 FastAPI Backend Development Bootcamp 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 FastAPI Backend Development Bootcamp 课程共包含 4 节课。

本课时的部分内容尚未翻译，以英文显示。

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
免费开始

用 AI 导师学习 FastAPI Backend Development Bootcamp — 免费

在浏览器中编写并运行真实代码，获得全天候 AI 导师的即时帮助，并在网页或应用中继续学习。

课程
21
课程
84
免费开始学习查看完整课程

常见问题解答

「异步图像与文档转换」课时是免费的吗？

是的 — 「异步图像与文档转换」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 FastAPI Backend Development Bootcamp 课程的其余内容，请升级到 CoddyKit PRO。 FastAPI Backend Development Bootcamp 课程共包含 4 节课。

「异步图像与文档转换」这节课中我会学到什么？

在后台工作进程中处理缩略图、大小调整和格式转换，降低请求延迟。 你通过在浏览器中直接运行的动手代码来练习 FastAPI Backend Development Bootcamp，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

学习 FastAPI Backend Development Bootcamp 需要有经验吗？

无需任何先前经验。CoddyKit 上的 FastAPI Backend Development Bootcamp 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。 这是第 4 节课，共 4 节。

「异步图像与文档转换」课时需要多长时间？

大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。

我能在这节 FastAPI Backend Development Bootcamp 课中编写并运行代码吗？

能。每节 FastAPI Backend Development Bootcamp 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。

此课程中的所有课时

  1. 多部分上传与内容验证
  2. 流式响应与范围请求
  3. 将存储卸载到兼容 S3 的存储桶
  4. 异步图像与文档转换
← 返回 FastAPI Backend Development Bootcamp