HTTPX — Async Support

Versi async httpx: perbedaannya dengan versi sinkron hanya beberapa kata kunci.

Perubahan dari versi sinkron

# Sinkron
with httpx.Client() as client:
    r = client.get(url)
    data = r.json()

# Async — tiga kata yang berubah
async with httpx.AsyncClient() as client:
    r = await client.get(url)
    data = r.json()          # json() tetap sinkron — datanya sudah di memori

Paralel — inti Fase 3

import asyncio
import httpx

async def ambil(client: httpx.AsyncClient, url: str) -> dict:
    r = await client.get(url)
    r.raise_for_status()
    return r.json()

async def main() -> None:
    async with httpx.AsyncClient(timeout=30.0) as client:
        hasil = await asyncio.gather(
            *(ambil(client, u) for u in urls),
            return_exceptions=True,
        )
    for u, h in zip(urls, hasil):
        if isinstance(h, Exception):
            print(f"gagal {u}: {h}")

asyncio.run(main())

Membatasi concurrency

sem = asyncio.Semaphore(10)

async def ambil_terbatas(client: httpx.AsyncClient, url: str) -> dict:
    async with sem:
        r = await client.get(url)
        r.raise_for_status()
        return r.json()

async def main() -> None:
    async with httpx.AsyncClient() as client:
        await asyncio.gather(*(ambil_terbatas(client, u) for u in urls))

Batas kedua ada di level koneksi:

limits = httpx.Limits(max_connections=20, max_keepalive_connections=10)
client = httpx.AsyncClient(limits=limits)

Bedanya: semaphore membatasi request logis, Limits membatasi koneksi TCP. Untuk mematuhi rate limit penyedia API, semaphore yang lebih tepat.

Client berumur panjang

Di aplikasi server, jangan membuat client per request. Buat sekali saat startup:

from contextlib import asynccontextmanager
from fastapi import FastAPI
import httpx

@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.http = httpx.AsyncClient(timeout=30.0)
    yield
    await app.state.http.aclose()

app = FastAPI(lifespan=lifespan)

@app.get("/proxy")
async def proxy():
    r = await app.state.http.get("https://api.contoh.com/data")
    return r.json()

Pola lifespan ini dibahas lengkap di Fase 5.

Streaming

async with client.stream("GET", "/besar") as r:
    r.raise_for_status()
    async for potongan in r.aiter_bytes():
        await tulis(potongan)

Server-Sent Events — bentuk yang dipakai API LLM:

import json

async with client.stream("POST", "/chat", json=payload) as r:
    r.raise_for_status()
    async for baris in r.aiter_lines():
        if not baris.startswith("data: "):
            continue
        muatan = baris[6:]
        if muatan == "[DONE]":
            break
        yield json.loads(muatan)

Pembatalan

try:
    async with asyncio.timeout(10):
        r = await client.get("https://lambat.contoh.com")
except TimeoutError:
    print("dibatalkan")     # koneksi ditutup otomatis

task = asyncio.create_task(client.get(url))
task.cancel()
try:
    await task
except asyncio.CancelledError:
    pass

httpx menghormati pembatalan asyncio: request yang sedang berjalan dihentikan dan koneksinya dikembalikan ke pool.

Testing async

import pytest, httpx

@pytest.mark.asyncio
async def test_ambil():
    def handler(request: httpx.Request) -> httpx.Response:
        return httpx.Response(200, json={"ok": True})

    async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client:
        hasil = await ambil(client, "https://apa-saja")
        assert hasil == {"ok": True}

uv add --dev pytest-asyncio

Checkpoint Fase 3

import asyncio, time
import httpx

URLS = ["https://httpbin.org/delay/1"] * 20
sem = asyncio.Semaphore(5)

async def ambil(client: httpx.AsyncClient, url: str) -> int:
    async with sem:
        r = await client.get(url, timeout=30.0)
        r.raise_for_status()
        return r.status_code

async def main() -> None:
    mulai = time.perf_counter()
    async with httpx.AsyncClient() as client:
        hasil = await asyncio.gather(
            *(ambil(client, u) for u in URLS), return_exceptions=True
        )
    berhasil = sum(1 for h in hasil if not isinstance(h, Exception))
    print(f"{berhasil}/{len(URLS)} sukses dalam {time.perf_counter() - mulai:.1f}s")

asyncio.run(main())

20 URL, concurrency 5, tiap request menunggu 1 detik → sekitar 4 detik. Versi sekuensialnya butuh 20 detik. Kamu harus bisa menjelaskan dari mana angka 4 itu datang.