HTTPX Snippets: 実践ガイド

1 HTTPX の紹介

HTTPX は Python のモダンな HTTP クライアントです。馴染みのある requests API を維持しつつ、HTTP/2、非同期サポート、コネクションプーリングを標準で提供します。requests を知っていれば、HTTPX のほとんどを理解したことになります。「同期か非同期か？」を心に留めておくだけです。

同じメンタルモデルで同期・非同期
モダンなプロトコル: HTTP/2 (利用可能な場合)
優れたデフォルト: コネクションプーリング、タイムアウト、リトライ

「次へ」をクリックして HTTPX をインストールし、最初のリクエストを送信しましょう。

2 インストール

仮想環境を使用してください。その後 HTTPX をインストールします。[http] extra を使用すると、標準トランスポートがインストールされます（追加の依存関係は不要です）。

pip install httpx[http]

HTTP/2 が必要な場合は、h2 もインストールしてください: pip install h2。macOS で HTTPS を使用する場合、証明書が必要になることがあります: pip install certifi と設定 SSL_CERT_FILE または REQUESTS_CA_BUNDLE。

3 簡単な GET リクエスト

これは最小限の同期 GET リクエストです。ステータスコードと抜粋の一部を出力します。

import httpx

with httpx.Client() as client:
    r = client.get("https://httpbin.org/get")
    print("status:", r.status_code)
    print("content[:120]:", r.text[:120])

クライアントはコンテキストマネージャーとして使用されるため、接続はきれいに解放されます。それだけで、「ファイルオープン数が多すぎます」という驚くほど多くのエラーを防ぐことができます。

4 ヘッダー、パラメータ、タイムアウト

ヘッダーとクエリパラメータは通常の辞書として渡します。常に明示的なタイムアウトを設定してください。

import httpx

with httpx.Client(
    base_url="https://httpbin.org",
    headers={"User-Agent": "httpx-snippets/1.0"},
    params={"page": "1", "per_page": "5"},
    timeout=5.0,
) as client:
    r = client.get("/get")
    r.raise_for_status()
    data = r.json()
    print(data.get("args"))

base_url を使用して重複を減らし、r.raise_for_status() を使用して HTTP エラーをサイレントフェイルではなく例外に変換します。

5 JSON で POST

JSON POST は一行で完了します。クライアントはシリアライズし、正しい Content-Type を設定します。

import httpx

payload = {"name": "Ada", "skills": ["math", "coding"]}

with httpx.Client(base_url="https://httpbin.org") as client:
    r = client.post("/post", json=payload)
    r.raise_for_status()
    print(r.json()["json"])

6 エラーハンドリング

可能な場合は特定の例外を使用し、それ以外の場合は基底クラスにフォールバックします。

import httpx
from httpx import ConnectTimeout, ReadTimeout, RemoteProtocolError

try:
    with httpx.Client(timeout=1.0) as client:
        r = client.get("https://httpbin.org/delay/2")
        r.raise_for_status()
except ConnectTimeout:
    print("接続できませんでした（ネットワークダウン、DNS、拒否）。")
except ReadTimeout:
    print("サーバーの応答に時間がかかりすぎました。")
except RemoteProtocolError:
    print("通信中に接続が切れました。")
except httpx.HTTPStatusError as e:
    print(f"HTTP エラー: {e.response.status_code} {e.response.reason_phrase}")
except httpx.RequestError as e:
    print(f"リクエストに失敗しました: {e}")

7 非同期リクエスト (asyncio)

並行処理が必要な場合は httpx.AsyncClient に切り替えます。I/O 呼び出しは asyncio.gather でグループ化します。

import asyncio
import httpx

async def fetch(url: str):
    async with httpx.AsyncClient(timeout=5.0) as client:
        r = await client.get(url)
        r.raise_for_status()
        return r.json()

async def main():
    urls = ["https://httpbin.org/get"] * 4
    results = await asyncio.gather(*(fetch(u) for u in urls))
    for i, data in enumerate(results, 1):
        print(i, data.get("origin"))

asyncio.run(main())

8 リトライとバックオフ

一時的な障害に対しては、httpx.Limits と Timeout を使用してリトライポリシーでトランスポートをラップします。

import httpx
from urllib3.util.retry import Retry
from httpx import LegacyHTTPTransport

retry = Retry(
    total=3,
    backoff_factor=0.5,
    status_forcelist=[429, 500, 502, 503, 504],
    allowed_methods=["GET", "POST", "PUT", "DELETE", "HEAD", "OPTIONS"],
)

limits = httpx.Limits(max_keepalive_connections=5, max_connections=10)
timeout = httpx.Timeout(5.0, connect=2.0)

transport = LegacyHTTPTransport(retries=retry)
client = httpx.Client(transport=transport, limits=limits, timeout=timeout)

r = client.get("https://httpbin.org/status/503")
print(r.status_code)

注意: リトライロジックはトランスポートレイヤーにあります。控えめにしてください。リトライは障害を増幅させる可能性があります。

9 高度なタイムアウト

接続タイムアウトと読み取りタイムアウトを分割して、より厳密な制御を行います。ストリーミング応答の場合は、読み取りタイムアウトを無効にすることもできます（None）。

import httpx

timeout = httpx.Timeout(connect=2.0, read=5.0, write=2.0, pool=1.0)
with httpx.Client(timeout=timeout) as client:
    r = client.get("https://httpbin.org/stream/3")  # ストリームライン
    for line in r.iter_lines():
        print(line)

10 ファイルのアップロード (Multipart)

files= でファイルを、data= で通常のフォームフィールドを送信します。

import httpx

with httpx.Client(base_url="https://httpbin.org") as client:
    with open("README.md", "rb") as f:
        files = {"file": ("README.md", f, "text/markdown")}
        data = {"uploader": "me"}
        r = client.post("/post", files=files, data=data)
        r.raise_for_status()
        print(r.json()["headers"].get("Content-Type"))

11 FastAPI で HTTPX を使用する

アプリ起動時に単一の再利用可能なクライアントを作成し、シャットダウン時に解放します。

from fastapi import FastAPI
import httpx

app = FastAPI()
client = httpx.Client(base_url="https://httpbin.org", timeout=5.0)

@app.on_event("startup")
async def on_startup():
    await client.__aenter__()

@app.on_event("shutdown")
async def on_shutdown():
    await client.__aexit__(None, None, None)

@app.get("/proxy/get")
async def proxy_get():
    r = await client.get("/get")
    r.raise_for_status()
    return r.json()

本番環境では、トランスポートに Limits とリトライポリシーをマウントします。アップストリームが頻繁に通信する場合は、一般的な応答をキャッシュします。