Python: Gọi nhiều API song song với asyncio.gather và Semaphore trong 10 phút

Khi cần gọi hàng chục API — đồng bộ dữ liệu, crawl giá, bắn webhook — vòng lặp for tuần tự cộng dồn thời gian chờ của từng request: 20 request × 1 giây = 20 giây. asyncio.gather chạy nhiều coroutine đồng thời trong một event loop, còn asyncio.Semaphore giới hạn số request bay cùng lúc để không làm đối tác chặn IP. Kỹ thuật này dùng được cho mọi tác vụ I/O-bound và chỉ cần thư viện httpx.

Vấn đề: vòng lặp tuần tự cộng dồn thời gian chờ

Giả sử cần lấy thông tin 20 user từ một API, mỗi request mất khoảng 1 giây. Cách "ngây thơ" nhất là dùng httpx.Client đồng bộ:

import time
import httpx

URLS = [f"https://api.example.com/users/{i}" for i in range(20)]

def fetch_all():
    results = []
    with httpx.Client(timeout=10) as client:
        for url in URLS:
            results.append(client.get(url).json())
    return results

t0 = time.perf_counter()
fetch_all()
print(f"finished in {time.perf_counter() - t0:.1f} seconds")

Code chạy đúng, nhưng tổng thời gian ≈ 20 giây vì mỗi request phải chờ request trước xong. CPU gần như ngồi chơi toàn bộ thời gian — nút thắt là chờ I/O, không phải xử lý. Với 200 bản ghi, bạn sẽ chờ hơn 3 phút.

Giải pháp: asyncio.gather + Semaphore

asyncio cho phép nhiều coroutine "tạm dừng" tại điểm chờ I/O và nhường event loop cho coroutine khác. asyncio.gather(*coros) gom chúng lại, chạy đồng thời và trả kết quả đúng thứ tự bạn truyền vào. Nhưng đừng bắn 200 request một lúc — API đối tác có thể rate-limit hoặc chặn IP. asyncio.Semaphore(n) là một "cổng" chỉ cho n coroutine đi qua cùng lúc:

import asyncio
import httpx

async def fetch_one(client, url, sem):
    async with sem:  # lấy vé trước khi gọi HTTP
        resp = await client.get(url, timeout=10)
        resp.raise_for_status()
        return resp.json()

async def fetch_all_async(urls):
    sem = asyncio.Semaphore(5)  # tối đa 5 request cùng lúc
    async with httpx.AsyncClient() as client:
        return await asyncio.gather(
            *(fetch_one(client, u, sem) for u in urls),
            return_exceptions=True,
        )

results = asyncio.run(fetch_all_async(URLS))

Cách vận hành: mỗi fetch_one lấy một "vé" từ semaphore trước khi gọi HTTP; tối đa 5 request bay cùng lúc, request nào xong thì request khác thế chỗ. Với 20 request × 1 giây và Semaphore(5), tổng thời gian giảm còn ≈ 4 giây thay vì 20. Đo trên server local (20 request × 0,3 giây): 6,19 giây → 1,41 giây.

Gom lỗi gọn với return_exceptions=True

Mặc định, một coroutine lỗi sẽ làm gather ném exception ngay và bạn mất kết quả của các request đã xong. Truyền return_exceptions=True thì lỗi quay về như một phần tử bình thường trong danh sách kết quả — lọc ra sau:

ok = [r for r in results if not isinstance(r, Exception)]
failed = [r for r in results if isinstance(r, Exception)]
print(f"success: {len(ok)}/{len(results)}")
for e in failed:
    print("failed request:", repr(e))

Mẫu này rất hợp cho job đồng bộ dữ liệu: 98/100 bản ghi vẫn xử lý được, 2 lỗi vào log để retry sau — thay vì cả batch chết vì một request hỏng.

Áp dụng vào dự án

Bắt đầu với Semaphore(5), rồi tăng dần trong lúc theo dõi thời gian phản hồi và mã 429 từ API đối tác. Luôn đặt timeout cho mọi request — một request treo mãi sẽ giữ "vé" semaphore mãi và kẹt cả batch. Nếu dự án đã dùng FastAPI, các endpoint async def có thể await fetch_all_async() trực tiếp mà không cần asyncio.run().

Bài viết rút gọn. Đọc bản đầy đủ (tiếng Anh) tại: https://linhfish.dev/en/2026/10/23/python-call-multiple-apis-in-parallel-with-asyncio-gather-and-semaphore-in-10-minutes/