Fase 5 · Bangun API dengan FastAPI

Concurrency and async / await

Keputusan async atau sync di FastAPI: satu aturan sederhana yang menyelamatkan throughput aplikasimu.

Sumber asli fastapi.tiangolo.com Resmi Rangkuman ~5 menit baca

Intisari

async def berjalan di event loop. def biasa dijalankan FastAPI di threadpool.
Bahaya terbesar: kode blocking di dalam async def — tidak ada yang menyelamatkanmu.
Kalau ragu dan fungsimu memanggil library sinkron, pakai def biasa. Itu selalu aman.
Untuk aplikasi LLM, hampir semuanya seharusnya async def — SDK-nya punya versi async.
Middleware dan dependency mengikuti aturan yang sama.

Pohon keputusan

Fungsi ini memanggil sesuatu yang lambat?
│
├── Tidak (hanya olah data di memori)
│   └── async def — apa saja boleh
│
└── Ya
    ├── Ada versi async-nya? (AsyncAnthropic, httpx.AsyncClient, asyncpg)
    │   └── async def + await          ← TERBAIK
    │
    └── Hanya ada versi sinkron
        ├── def biasa                  ← aman, FastAPI pakai threadpool
        └── async def + asyncio.to_thread(...)   ← juga benar

Tiga kemungkinan

# ✅ BENAR — async penuh
@app.post("/chat")
async def chat(req: ChatRequest) -> ChatResponse:
    r = await async_client.messages.create(...)
    return ChatResponse(...)

# ✅ BENAR — sync, dijalankan di threadpool
@app.post("/proses")
def proses(req: ProsesRequest) -> Hasil:
    return library_sinkron.hitung(req.data)

# ❌ SALAH — blocking di dalam async
@app.post("/salah")
async def salah(req: Request) -> dict:
    return requests.get("https://api.contoh.com").json()

Kenapa yang ketiga begitu merusak: event loop berjalan di satu thread. Selama requests.get() menunggu jaringan, tidak ada request lain yang bisa dilayani — bahkan request ke endpoint yang sama sekali berbeda. Satu baris ini mengubah server dengan ribuan koneksi jadi server yang melayani satu per satu.

Bagaimana FastAPI memperlakukan keduanya

	`async def`	`def`
Dijalankan di	Event loop	Threadpool
Concurrency	Ribuan	Terbatas ukuran threadpool (~40)
Overhead	Rendah	Perpindahan thread
Kalau blocking	Membekukan semuanya	Hanya menempati satu thread
Bisa `await`	Ya	Tidak

Daftar periksa blocking

Blocking	Ganti dengan
`requests.get()`	`await httpx_client.get()`
`anthropic.Anthropic()`	`anthropic.AsyncAnthropic()`
`time.sleep()`	`await asyncio.sleep()`
Driver DB sinkron	`asyncpg`, `aiosqlite`, atau `def` biasa
`boto3`	`await asyncio.to_thread(...)` atau `aioboto3`
Baca file besar	`await asyncio.to_thread(p.read_text)`
Komputasi berat (numpy)	`def` biasa, atau process pool

import asyncio

@app.post("/embed")
async def embed(req: EmbedRequest) -> EmbedResponse:
    vektor = await asyncio.to_thread(model_embedding.encode, req.teks)
    return EmbedResponse(vektor=vektor.tolist())

Dependency mengikuti aturan yang sama

async def get_db():
    async with async_session() as s:
        yield s

def get_config():
    return baca_config_sinkron()      # dijalankan di threadpool

Middleware

import time
from fastapi import Request

@app.middleware("http")
async def catat_waktu(request: Request, call_next):
    mulai = time.perf_counter()
    response = await call_next(request)
    durasi = time.perf_counter() - mulai
    response.headers["X-Process-Time"] = f"{durasi:.3f}"
    logger.info("%s %s %.3fs", request.method, request.url.path, durasi)
    return response

Middleware berjalan untuk SETIAP request. Satu operasi blocking di sini akan memperlambat seluruh aplikasi secara merata — dan sulit dilacak karena tidak menempel pada endpoint tertentu.

Menemukan blocking yang tersembunyi

1. Debug mode asyncio

PYTHONASYNCIODEBUG=1 uv run uvicorn myapp.main:app

Memberi peringatan untuk callback yang berjalan lebih dari 100 ms.

2. Ruff

[tool.ruff.lint]
select = ["ASYNC"]

Mendeteksi time.sleep, requests, dan operasi file sinkron di dalam fungsi async.

3. Uji beban sederhana

# Endpoint yang hanya menunggu 1 detik
@app.get("/uji")
async def uji():
    await asyncio.sleep(1)
    return {"ok": True}

ab -n 100 -c 50 http://localhost:8000/uji

Kalau async-mu benar, 100 request dengan concurrency 50 selesai dalam ~2 detik. Kalau jauh lebih lama, ada yang blocking di jalur request.

Ukuran threadpool

import anyio

@app.on_event("startup")
async def perbesar_threadpool():
    limiter = anyio.to_thread.current_default_thread_limiter()
    limiter.total_tokens = 100      # default sekitar 40

Naikkan hanya kalau kamu memang banyak memakai endpoint def biasa yang lambat. Solusi yang lebih baik biasanya membuatnya benar-benar async.

Untuk aplikasi LLM

Hampir semua yang dilakukan aplikasi LLM adalah I/O: panggilan API, query vector store, tulis database. Semuanya punya versi async. Jadi jawabannya hampir selalu: async penuh dari ujung ke ujung.

client = anthropic.AsyncAnthropic()       # bukan anthropic.Anthropic()
http = httpx.AsyncClient()                # bukan httpx.Client()

@app.post("/chat")
async def chat(req: ChatRequest) -> ChatResponse:
    dok = await vector_store.cari(req.pesan)          # async
    r = await client.messages.create(...)             # async
    await db.simpan(req.session_id, r)                # async
    return ChatResponse(...)

Rangkuman ini sengaja dipangkas ke bagian yang dipakai di roadmap. Buka sumber aslinya saat kamu butuh detail lengkap atau referensi parameter.