Transformasi Gambar dan Dokumen Asinkron
Proses gambar mini, pengubahan ukuran, dan konversi format di pekerja latar belakang agar latensi permintaan tetap rendah.
Transformasi Gambar dan Dokumen Asinkron adalah pelajaran FastAPI Backend Development Bootcamp gratis di CoddyKit. Ini adalah pelajaran 4 dari 4. Kamu bisa membaca pelajaran lengkapnya di bawah secara gratis — lalu praktikkan langsung di browser dengan editor kode bawaan dan tutor AI 24/7. Ini adalah bagian dari jalur belajar FastAPI Backend Development Bootcamp, dan progresmu tersinkronisasi di web dan aplikasi CoddyKit. Kursus FastAPI Backend Development Bootcamp mencakup 4 pelajaran total.
Bagian dari pelajaran ini belum diterjemahkan dan ditampilkan dalam bahasa Inggris.
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 Acceptedwith 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_idthe 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_forso 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 NonePolling 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 FalseQuick 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
202with a job id, enqueue the work - Use a real queue (Celery/RQ/arq + Redis) for CPU-heavy transforms; reserve
BackgroundTasksfor 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
Belajar FastAPI Backend Development Bootcamp dengan tutor AI — gratis
Tulis dan jalankan kode asli di browser kamu, dapatkan bantuan instan dari tutor AI 24/7, dan lanjutkan di mana kamu tinggalkan di web atau aplikasi.
- Kursus
- 21
- Pelajaran
- 84
Pertanyaan yang Sering Diajukan
Apakah pelajaran “Transformasi Gambar dan Dokumen Asinkron” gratis?
Ya — teks lengkap “Transformasi Gambar dan Dokumen Asinkron” gratis dibaca di sini di web. Untuk praktiknya secara interaktif (editor kode bawaan dan tutor AI 24/7) dan buka sisa kursus FastAPI Backend Development Bootcamp, upgrade ke CoddyKit PRO. Kursus FastAPI Backend Development Bootcamp mencakup 4 pelajaran total.
Apa yang akan aku pelajari di “Transformasi Gambar dan Dokumen Asinkron”?
Proses gambar mini, pengubahan ukuran, dan konversi format di pekerja latar belakang agar latensi permintaan tetap rendah. Kamu berlatih FastAPI Backend Development Bootcamp dengan kode praktik yang langsung kamu jalankan di browser, dan tutor AI 24/7 menjawab pertanyaanmu saat kamu mengerjakan pelajaran ini.
Apakah aku perlu pengalaman untuk memulai FastAPI Backend Development Bootcamp?
Tidak diperlukan pengalaman sebelumnya. FastAPI Backend Development Bootcamp di CoddyKit dirancang untuk pemula hingga pelajar tingkat lanjut, jadi kamu bisa memulai di sini atau dari awal dan belajar sesuai kecepatan kamu sendiri. Ini adalah pelajaran 4 dari 4.
Berapa lama pelajaran “Transformasi Gambar dan Dokumen Asinkron” memakan waktu?
Sebagian besar pelajaran CoddyKit memakan waktu sekitar 5–10 menit. Setiap pelajaran ringkas dan interaktif, jadi kamu membuat kemajuan stabil dan melanjutkan dari tempat kamu tinggalkan di web dan aplikasi.
Bisakah aku menulis dan menjalankan kode dalam pelajaran FastAPI Backend Development Bootcamp ini?
Ya. Setiap pelajaran FastAPI Backend Development Bootcamp menyertakan editor kode bawaan, jadi kamu menulis dan menjalankan kode nyata langsung di browser dan mendapatkan umpan balik AI instan — tidak diperlukan penyiapan lokal.
Semua pelajaran dalam kursus ini
- Unggahan Multipart dan Validasi Konten
- Respons Streaming dan Permintaan Rentang
- Memindahkan Penyimpanan ke Bucket yang Kompatibel dengan S3
- Transformasi Gambar dan Dokumen Asinkron