Gọi API trong Python với requests và httpx

Gọi API là công việc gần như dự án nào cũng đụng tới. Bài này đi qua requests — thư viện quen thuộc nhất — và httpx, lựa chọn hiện đại hỗ trợ cả bất đồng bộ.

requests — đơn giản và phổ biến

pip install requests

import requests

r = requests.get("https://api.github.com/users/python")
print(r.status_code)
print(r.json()["name"])

Các phương thức HTTP

requests.get(url, params={"trang": 2, "gioi_han": 50})
requests.post(url, json={"ten": "Minh", "tuoi": 30})
requests.put(url, json=du_lieu)
requests.patch(url, json={"tuoi": 31})
requests.delete(url)

Dùng tham số json= thay vì data=json.dumps(...) — thư viện tự đặt header Content-Type: application/json giúp bạn.

Kiểm tra lỗi

Đây là điều rất nhiều người bỏ qua: requests không tự ném ngoại lệ khi máy chủ trả về mã lỗi.

r = requests.get("https://api.example.com/khong-ton-tai")
print(r.status_code)     # 404
du_lieu = r.json()       # lỗi khó hiểu ở đây, không phải ở dòng trên

Hãy luôn kiểm tra:

r = requests.get(url)
r.raise_for_status()     # ném HTTPError nếu mã 4xx hoặc 5xx
du_lieu = r.json()

Timeout — đừng bao giờ quên

Mặc định requests chờ vô thời hạn. Nếu máy chủ đối tác treo, chương trình của bạn treo theo:

r = requests.get(url, timeout=10)              # tổng 10 giây
r = requests.get(url, timeout=(3.05, 27))      # (kết nối, đọc)

Đây là nguyên nhân phổ biến của những sự cố kiểu “dịch vụ đột nhiên đứng hình” trong môi trường thật.

Session — tái sử dụng kết nối

Khi gọi nhiều request tới cùng một máy chủ, dùng Session để tái sử dụng kết nối TCP:

with requests.Session() as s:
    s.headers.update({"Authorization": "Bearer token-cua-ban"})

    for trang in range(1, 11):
        r = s.get("https://api.example.com/items", params={"page": trang})
        r.raise_for_status()
        xu_ly(r.json())

Cách này nhanh hơn rõ rệt so với gọi requests.get mười lần, vì tránh được việc bắt tay TLS lặp lại.

Tự động thử lại

from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

s = requests.Session()
retry = Retry(
    total=3,
    backoff_factor=1,                       # chờ 1s, 2s, 4s
    status_forcelist=[429, 500, 502, 503, 504],
    allowed_methods=["GET", "POST"],
)
s.mount("https://", HTTPAdapter(max_retries=retry))

Chỉ thử lại với những mã lỗi có khả năng tự khỏi. Thử lại một lỗi 400 (dữ liệu gửi lên sai) là vô ích — lần sau cũng sẽ sai như vậy.

httpx — hiện đại hơn

pip install httpx

API của httpx gần như giống hệt requests, nên chuyển đổi rất dễ:

import httpx

r = httpx.get("https://api.github.com/users/python")
print(r.json()["name"])

Khác biệt lớn nằm ở hỗ trợ bất đồng bộ và HTTP/2:

import asyncio
import httpx

async def lay_du_lieu(client, url):
    r = await client.get(url)
    r.raise_for_status()
    return r.json()

async def main():
    urls = [f"https://api.example.com/items/{i}" for i in range(1, 21)]

    async with httpx.AsyncClient(timeout=10) as client:
        ket_qua = await asyncio.gather(
            *(lay_du_lieu(client, u) for u in urls)
        )
    return ket_qua

asyncio.run(main())

Hai mươi request trên chạy song song. Với requests tuần tự, tổng thời gian là tổng của từng request; ở đây, nó xấp xỉ thời gian của request chậm nhất.

Giới hạn số request đồng thời

Đừng bắn hàng nghìn request cùng lúc — bạn sẽ bị chặn hoặc làm quá tải máy chủ đối tác:

async def main():
    gioi_han = asyncio.Semaphore(10)     # tối đa 10 request cùng lúc

    async def lay_co_gioi_han(client, url):
        async with gioi_han:
            return await lay_du_lieu(client, url)

    async with httpx.AsyncClient() as client:
        return await asyncio.gather(
            *(lay_co_gioi_han(client, u) for u in urls)
        )

Xử lý lỗi đầy đủ

import httpx

try:
    r = httpx.get(url, timeout=10)
    r.raise_for_status()
    du_lieu = r.json()
except httpx.TimeoutException:
    print("Máy chủ phản hồi quá chậm")
except httpx.HTTPStatusError as e:
    print(f"Máy chủ trả về lỗi {e.response.status_code}")
except httpx.RequestError as e:
    print(f"Không kết nối được: {e}")
except ValueError:
    print("Phản hồi không phải JSON hợp lệ")

Bảo mật khi gọi API

Đừng đưa khóa API vào mã nguồn. Đọc từ biến môi trường:

import os

khoa = os.environ["API_KEY"]      # lỗi ngay nếu chưa cấu hình
headers = {"Authorization": f"Bearer {khoa}"}

Dùng os.environ[...] thay vì os.environ.get(...) ở đây là có chủ ý: nếu thiếu cấu hình, chương trình dừng ngay với thông báo rõ ràng, thay vì âm thầm gửi request thiếu xác thực rồi báo lỗi khó hiểu.

Và đừng bao giờ tắt kiểm tra chứng chỉ TLS:

requests.get(url, verify=False)     # đừng làm thế này

Dòng trên vô hiệu hóa toàn bộ lớp bảo vệ chống tấn công xen giữa. Nếu gặp lỗi chứng chỉ, hãy sửa gốc rễ vấn đề — cập nhật kho chứng chỉ hoặc chỉ định file CA đúng — thay vì bỏ qua kiểm tra.