Fase 3 · Async & HTTP

Developing with asyncio

Kesalahan async yang paling sering terjadi dan cara mendeteksinya sebelum masuk production.

Sumber asli docs.python.org Resmi Rangkuman ~6 menit baca

Intisari

PYTHONASYNCIODEBUG=1 menyalakan peringatan untuk coroutine yang lupa di-await dan task lambat.
Satu panggilan blocking di dalam async def membekukan seluruh aplikasi.
Pola yang berbahaya: requests.get(), time.sleep(), query DB sinkron, open().read() file besar.
Exception di dalam task yang tidak pernah di-await hilang tanpa jejak.
asyncio tidak thread-safe. Untuk memanggil dari thread lain, pakai run_coroutine_threadsafe.

Nyalakan debug mode saat mengembangkan

PYTHONASYNCIODEBUG=1 uv run main.py

asyncio.run(main(), debug=True)

Debug mode memberi tahu kamu tentang:

Coroutine yang tidak pernah di-await
Callback yang berjalan lebih dari 100 ms (indikasi blocking)
Task yang dihancurkan saat masih menunggu
Exception yang tidak pernah diambil

Jebakan #1: panggilan blocking

Ini kesalahan async yang paling merusak, dan paling sulit dilihat karena kodenya tampak benar:

async def ambil_data(url: str) -> dict:
    return requests.get(url).json()      # ← MEMBEKUKAN SELURUH EVENT LOOP

Kenapa ini fatal: event loop berjalan di satu thread. Selama requests.get() menunggu jaringan, tidak ada task lain yang bisa berjalan — termasuk request dari pengguna yang sama sekali tidak berhubungan. Satu baris ini mengubah server async menjadi server yang melayani satu request pada satu waktu.

Daftar pola blocking yang sering lolos

Blocking	Ganti dengan
`requests.get()`	`await httpx_client.get()`
`time.sleep(n)`	`await asyncio.sleep(n)`
`open(p).read()` (file besar)	`await asyncio.to_thread(p.read_text)`
Query DB sinkron	Driver async, atau `to_thread`
Komputasi CPU berat	`run_in_executor` dengan process pool
`subprocess.run()`	`asyncio.create_subprocess_exec()`
Client SDK versi sinkron	Versi async-nya (`AsyncAnthropic`)

Membungkus kode blocking yang tidak bisa dihindari

hasil = await asyncio.to_thread(fungsi_blocking, arg)

Ruff bisa menangkap sebagian besar ini. Aktifkan aturan ASYNC (flake8-async) di pyproject.toml — ia mendeteksi time.sleep, requests, dan operasi file sinkron di dalam fungsi async.

Jebakan #2: lupa `await`

async def main():
    ambil("http://a")           # ← tidak pernah jalan
    print("selesai")

RuntimeWarning: coroutine 'ambil' was never awaited

Varian yang lebih halus — sering muncul di comprehension:

hasil = [ambil(u) for u in urls]           # ❌ list berisi coroutine, bukan hasil
hasil = await asyncio.gather(*(ambil(u) for u in urls))   # ✅

Jebakan #3: exception yang hilang

async def main():
    asyncio.create_task(mungkin_gagal())    # ❌ tidak dipegang, tidak di-await
    await asyncio.sleep(10)
    # Kalau task itu error, kamu tidak akan pernah tahu

# ✅ TaskGroup menyalurkan semua exception ke luar
async with asyncio.TaskGroup() as tg:
    tg.create_task(mungkin_gagal())

# ✅ atau simpan referensinya dan tangani sendiri
task = asyncio.create_task(mungkin_gagal())
task.add_done_callback(lambda t: t.exception() and logger.error("gagal: %s", t.exception()))

Jebakan #4: task yang dibuang garbage collector

# ❌ event loop hanya memegang referensi lemah
asyncio.create_task(latar())

# ✅ pegang referensinya sampai selesai
_tasks: set[asyncio.Task] = set()

t = asyncio.create_task(latar())
_tasks.add(t)
t.add_done_callback(_tasks.discard)

Aturan Ruff RUF006 mendeteksi pola ini.

Jebakan #5: asyncio bukan thread-safe

# ❌ dari thread lain
loop.call_soon(callback)

# ✅ versi yang aman lintas thread
loop.call_soon_threadsafe(callback)
future = asyncio.run_coroutine_threadsafe(coro, loop)
hasil = future.result(timeout=30)

Jebakan #6: menutup resource

# ❌ client tidak pernah ditutup — koneksi bocor
client = httpx.AsyncClient()
await client.get(url)

# ✅ context manager
async with httpx.AsyncClient() as client:
    await client.get(url)

# ✅ untuk client berumur panjang (lihat lifespan FastAPI di Fase 5)
client = httpx.AsyncClient()
try:
    ...
finally:
    await client.aclose()

Logging yang membantu debugging async

import logging

logging.basicConfig(
    level=logging.DEBUG,
    format="%(asctime)s %(name)s %(levelname)s [%(taskName)s] %(message)s",
)

%(taskName)s (Python 3.12+) mencantumkan nama task di setiap baris log — sangat membantu saat puluhan operasi berjalan bersamaan dan lognya bercampur.

asyncio.create_task(ambil(u), name=f"ambil-{u}")   # beri nama supaya terbaca di log

Melihat apa yang sedang berjalan

for t in asyncio.all_tasks():
    print(t.get_name(), t.get_coro())

Kapan async justru merugikan

Situasi	Async?	Alasan
Banyak panggilan API / LLM	✅ Ya	I/O-bound — inilah kasus idealnya
Server web dengan banyak koneksi	✅ Ya	Ribuan koneksi menganggur, satu thread
Streaming respons ke browser	✅ Ya	Koneksi lama, data sedikit-sedikit
Komputasi berat (matematika, gambar)	❌ Tidak	CPU-bound — pakai `multiprocessing`
Script sekali jalan, satu panggilan	❌ Tidak	Kompleksitas tanpa manfaat

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