次世代Python HTTPクライアント: httpx 詳細解説

🚀 httpx とは？

httpx は、Python 3 用のモダンで高機能なHTTPクライアントライブラリです。有名な requests ライブラリにインスパイアされた使いやすいAPIを持ちながら、現代的なWeb開発の要求に応えるための強力な機能を備えています。

主な特徴として、非同期 (async/await) リクエストのサポートと、HTTP/1.1 および HTTP/2 の両プロトコルへの対応が挙げられます。これにより、大量のリクエストを効率的に処理したり、最新のWebサーバーとの通信を最適化したりすることが可能です。

requests を使ったことがある開発者であれば、比較的スムーズに httpx へ移行できます。それでいて、より高度な機能やパフォーマンスを求める場合に強力な選択肢となります。

💻 インストール

httpx のインストールは pip を使って簡単に行えます。

pip install httpx

HTTP/2 サポートを利用したい場合は、追加の依存関係を含めてインストールします。Brotli圧縮のサポートも同様に追加できます。

# HTTP/2 サポートを追加
pip install httpx[http2]

# HTTP/3 サポートを追加 (httpcore の機能)
# httpx は内部で httpcore を使用しており、httpcore が HTTP/3 をサポートします
pip install httpx[http3]

# Brotli 圧縮サポートを追加
pip install httpx[brotli]

# 複数まとめてインストール
pip install httpx[http2,http3,brotli]

また、コマンドラインから httpx を使いたい場合は、cli オプションを追加します。

pip install httpx[cli]

注意: httpx は Python 3.8 以降が必要です。

⚙️ 基本的な使い方 (同期クライアント)

requests ライブラリと同様の感覚で、同期的なHTTPリクエストを簡単に送信できます。

GET リクエスト

最も基本的な GET リクエストの例です。

import httpx

# シンプルな GET リクエスト
response = httpx.get('https://httpbin.org/get')

# ステータスコードの確認
print(f"ステータスコード: {response.status_code}")

# ヘッダーの表示 (辞書ライクなオブジェクト)
print(f"Content-Type: {response.headers.get('content-type')}")

# レスポンスボディ (テキスト)
# print(f"レスポンスボディ:\n{response.text}")

# レスポンスボディ (JSON)
try:
    json_data = response.json()
    print("レスポンス JSON:")
    import json
    print(json.dumps(json_data, indent=2, ensure_ascii=False))
except httpx.ResponseNotRead:
    print("レスポンスボディはまだ読み込まれていません。")
except json.JSONDecodeError:
    print("レスポンスは有効な JSON ではありません。")
except Exception as e:
    print(f"エラーが発生しました: {e}")

# レスポンスボディ (バイナリ)
# print(f"レスポンスボディ (bytes): {response.content}")

POST リクエスト (JSONデータ)

JSON データを送信する場合は json 引数を使用します。

import httpx
import json

payload = {'name': 'Taro Yamada', 'age': 30}

# POST リクエスト (JSON)
response = httpx.post('https://httpbin.org/post', json=payload)

print(f"ステータスコード: {response.status_code}")
try:
    response_data = response.json()
    print("サーバーからのレスポンス JSON:")
    print(json.dumps(response_data, indent=2, ensure_ascii=False))
except Exception as e:
    print(f"JSON デコードエラー: {e}")

クエリパラメータとカスタムヘッダー

params 引数でクエリパラメータを、headers 引数でカスタムヘッダーを付与できます。

import httpx
import json

params = {'search': 'python httpx', 'page': 1}
headers = {'User-Agent': 'my-cool-app/1.0', 'Accept': 'application/json'}

response = httpx.get('https://httpbin.org/get', params=params, headers=headers)

print(f"ステータスコード: {response.status_code}")
try:
    print("サーバーからのレスポンス JSON:")
    print(json.dumps(response.json(), indent=2, ensure_ascii=False))
except Exception as e:
    print(f"JSON デコードエラー: {e}")

Client インスタンスの使用

複数のリクエストで設定（ヘッダー、Cookie、タイムアウトなど）を共有したい場合や、コネクションプーリングを活用したい場合は httpx.Client を使用します。with 文を使うことで、リソースが確実に解放されます。

import httpx
import json

# Client インスタンスを作成 (with 文で自動的にクローズ)
with httpx.Client(base_url='https://httpbin.org', headers={'X-Custom-Header': 'value'}) as client:
    # GET リクエスト (/get)
    response_get = client.get('/get', params={'param1': 'test'})
    print(f"GET ステータス: {response_get.status_code}")
    try:
        print("GET レスポンス:")
        print(json.dumps(response_get.json(), indent=2, ensure_ascii=False))
    except Exception as e:
        print(f"JSON デコードエラー: {e}")

    print("-" * 20)

    # POST リクエスト (/post)
    response_post = client.post('/post', json={'message': 'Hello from Client!'})
    print(f"POST ステータス: {response_post.status_code}")
    try:
        print("POST レスポンス:")
        print(json.dumps(response_post.json(), indent=2, ensure_ascii=False))
    except Exception as e:
        print(f"JSON デコードエラー: {e}")

httpx.Client() は requests.Session() と同様の目的で使用され、コネクションの再利用や設定の共有に役立ちます。

⚡ 非同期クライアントの使い方 (Async)

httpx の真価が発揮されるのが非同期処理です。Python の asyncio などの非同期フレームワークと組み合わせることで、特に多数の I/O 待ちが発生するようなタスク（例: 大量の API リクエスト）を非常に効率的に実行できます。

非同期処理には async と await キーワード、そして httpx.AsyncClient を使用します。

基本的な非同期 GET リクエスト

import httpx
import asyncio
import json

async def fetch_data(url: str):
    # 非同期クライアントを生成
    async with httpx.AsyncClient() as client:
        print(f"リクエスト開始: {url}")
        # await を使って非同期にリクエストを実行
        response = await client.get(url)
        print(f"リクエスト完了: {url}, ステータス: {response.status_code}")
        try:
            return response.json()
        except Exception as e:
            print(f"JSON デコードエラー ({url}): {e}")
            return None

async def main():
    url1 = 'https://httpbin.org/get?source=async1'
    url2 = 'https://httpbin.org/delay/1' # 1秒待つAPI

    # asyncio.gather を使って複数の非同期タスクを並行実行
    results = await asyncio.gather(
        fetch_data(url1),
        fetch_data(url2)
    )

    print("\\n--- 取得結果 ---")
    for i, result in enumerate(results):
        if result:
            print(f"結果 {i+1}:")
            print(json.dumps(result, indent=2, ensure_ascii=False))
        else:
            print(f"結果 {i+1}: 取得失敗")

# 非同期関数を実行
if __name__ == "__main__":
    asyncio.run(main())

上記の例では、fetch_data という非同期関数を定義し、asyncio.gather を使って2つのURLへのリクエストを同時に（並行して）開始します。/delay/1 エンドポイントは1秒待ってから応答しますが、その間にもう一方の /get リクエストは処理を進めることができます。これにより、逐次的にリクエストを送るよりも全体の処理時間を短縮できます。

🤔 httpx vs requests 比較

httpx は多くの点で requests と似ていますが、重要な違いもいくつかあります。どちらを選択するかは、プロジェクトの要件によって異なります。

どちらを選ぶべきか？

• 単純なスクリプトや、同期処理のみで十分な場合: requests が依然としてシンプルで良い選択肢です。ドキュメントやコミュニティサポートも豊富です。
• 非同期処理が必要な場合（Webフレームワーク、大量のAPIアクセスなど）: httpx が最適です。
• HTTP/2 や HTTP/3 の機能を利用したい場合: httpx を選択します。
• モダンなPython機能（型ヒントなど）を最大限活用したい場合: httpx の方がより現代的な設計になっています。
• 新しいライブラリを試してみたい、より高機能・高性能なクライアントが必要な場合: httpx を検討する価値があります。

2025年現在、多くの新しいプロジェクト、特に非同期処理を前提とする FastAPI や Starlette などのフレームワークと組み合わせる際には、httpx が第一候補となるケースが増えています。

🛠️ 高度な機能

httpx は基本的なリクエスト以外にも、多くの高度な機能を提供します。

HTTP/2 と HTTP/3 サポート 🚀

前述の通り、httpx は HTTP/2 と HTTP/3 をサポートします。これにより、特に多重化（一つの接続で複数のリクエスト/レスポンスをやり取りする）によるパフォーマンス向上が期待できます。

利用するには、まず対応する依存ライブラリをインストールします。

# HTTP/2 の場合
pip install httpx[http2]

# HTTP/3 の場合
pip install httpx[http3]

そして、Client または AsyncClient を初期化する際に http2=True を指定します。HTTP/3 は、サーバーが対応していれば自動的にネゴシエーションされる場合があります（httpcore の実装に依存）。

import httpx
import asyncio

async def check_http_version(url: str):
    # http2=True を指定して HTTP/2 を優先的に試みる
    # サーバーが HTTP/3 に対応していれば、httpcore がそれを選択する可能性もある
    async with httpx.AsyncClient(http2=True, verify=False) as client: # verify=False は自己署名証明書などの場合に注意して使用
        try:
            response = await client.get(url)
            print(f"{url} - 使用されたプロトコル: {response.http_version}")
            # response.raise_for_status() # 必要に応じてステータスチェック
        except httpx.RequestError as exc:
            print(f"{url} へのリクエスト中にエラーが発生しました: {exc.__class__.__name__}")
        except Exception as e:
            print(f"{url} で予期せぬエラー: {e}")

async def main():
    # HTTP/2 や HTTP/3 をサポートしている可能性のあるサイト
    # 注意: サーバー側の設定やネットワーク環境により結果は異なります
    urls = [
        "https://www.google.com",
        "https://httpbin.org/headers", # HTTP/1.1 のみの場合が多い
        "https://cloudflare.com",
        # "https://localhost:8443" # ローカルでテストサーバーを立てる場合など
    ]
    tasks = [check_http_version(url) for url in urls]
    await asyncio.gather(*tasks)

if __name__ == "__main__":
    # 注意: 自己署名証明書などを使用しているサーバーの場合、SSL検証エラーが発生することがあります。
    # その場合は `verify=False` を指定するか、適切な証明書設定が必要です。
    # 本番環境での `verify=False` の使用はセキュリティリスクを伴います。
    asyncio.run(main())

HTTP/2 や HTTP/3 の利用はサーバー側の対応が必要です。また、HTTP/3 は UDP 上の QUIC プロトコルを使用するため、ネットワーク環境（ファイアウォールなど）によっては通信できない場合があります。HTTP/3 はデフォルトで TLSv1.3 による暗号化が必須です。

タイムアウト設定 ⏱️

httpx では、きめ細やかなタイムアウト設定が可能です。httpx.Timeout オブジェクトを使って指定します。

• connect: サーバーへの接続確立までのタイムアウト（秒）。
• read: サーバーからデータのチャンクを読み取る間のタイムアウト（秒）。
• write: サーバーにデータのチャンクを書き込む間のタイムアウト（秒）。
• pool: コネクションプールから接続を取得するまでのタイムアウト（秒）。

デフォルトではすべて5秒に設定されています。

import httpx

# 接続タイムアウトを 2秒、読み取りタイムアウトを 10秒に設定
timeout_config = httpx.Timeout(2.0, read=10.0)

try:
    with httpx.Client(timeout=timeout_config) as client:
        # このリクエストは最大10秒待機する
        response = client.get('https://httpbin.org/delay/5') # 5秒遅延するエンドポイント
        print("タイムアウトせずに成功しました。")
        # response = client.get('https://httpbin.org/delay/15') # これは read タイムアウトになるはず
        # print("タイムアウトせずに成功しました。")

except httpx.ReadTimeout:
    print("読み取りタイムアウトが発生しました。")
except httpx.ConnectTimeout:
    print("接続タイムアウトが発生しました。")
except httpx.TimeoutException as e:
    print(f"その他のタイムアウト関連エラー: {e}")
except httpx.RequestError as e:
    print(f"リクエストエラー: {e}")

リダイレクト制御

デフォルトでは、httpx はリダイレクトに追従します (最大30回)。follow_redirects=False を設定することで、リダイレクトを無効にできます。

import httpx

url = 'https://httpbin.org/redirect-to?url=https://www.example.com'

# リダイレクトに追従する (デフォルト)
response_follow = httpx.get(url)
print(f"追従あり - 最終URL: {response_follow.url}")
print(f"追従あり - ステータス: {response_follow.status_code}")
print(f"追従あり - リダイレクト履歴: {response_follow.history}")

print("\\n" + "-"*20 + "\\n")

# リダイレクトに追従しない
response_nofollow = httpx.get(url, follow_redirects=False)
print(f"追従なし - URL: {response_nofollow.url}")
print(f"追従なし - ステータス: {response_nofollow.status_code}") # 3xx 系になるはず
print(f"追従なし - Locationヘッダ: {response_nofollow.headers.get('location')}")
print(f"追従なし - リダイレクト履歴: {response_nofollow.history}") # 空になるはず

認証 (Authentication) 🔑

Basic 認証や Digest 認証などをサポートしています。auth 引数にタプルまたは認証クラスのインスタンスを渡します。

import httpx

# Basic 認証
basic_auth = ('your_username', 'your_password')
url_basic = 'https://httpbin.org/basic-auth/your_username/your_password'

try:
    response_basic = httpx.get(url_basic, auth=basic_auth)
    response_basic.raise_for_status() # エラーがあれば例外を発生させる
    print("Basic認証 成功")
    # print(response_basic.json())
except httpx.HTTPStatusError as e:
    print(f"Basic認証 失敗: ステータス {e.response.status_code}")
except httpx.RequestError as e:
    print(f"Basic認証 リクエストエラー: {e}")


# 独自認証クラスの例 (例: トークン認証)
class TokenAuth(httpx.Auth):
    def __init__(self, token):
        self._token = token

    def auth_flow(self, request):
        request.headers['Authorization'] = f'Bearer {self._token}'
        yield request

# custom_auth = TokenAuth('my-secret-token')
# response_custom = httpx.get('https://api.example.com/protected', auth=custom_auth)
# print(f"カスタム認証 ステータス: {response_custom.status_code}")

クライアントのカスタマイズ

httpx.Client や httpx.AsyncClient をインスタンス化する際に、様々なオプションを指定して動作をカスタマイズできます。

• base_url: ベースURL。相対パスでリクエストを送る際に便利。
• headers: すべてのリクエストに適用されるデフォルトヘッダー。
• cookies: すべてのリクエストに適用されるデフォルトクッキー。
• params: すべてのリクエストに適用されるデフォルトクエリパラメータ。
• timeout: デフォルトのタイムアウト設定 (httpx.Timeout オブジェクト)。
• limits: 接続数の制限 (httpx.Limits オブジェクト)。同時接続数やキープアライブ接続数を制御。
• proxies: プロキシ設定。
• verify: SSL/TLS 証明書の検証設定 (True, False, または証明書ファイルのパス)。
• cert: クライアント証明書の設定。
• http1, http2: 使用するHTTPバージョンを制御。
• transport: カスタムトランスポートクラスのインスタンス。
• event_hooks: イベントフック関数。
• follow_redirects: デフォルトのリダイレクト追従設定。

import httpx

# カスタマイズ例
limits = httpx.Limits(max_connections=100, max_keepalive_connections=20)
timeout = httpx.Timeout(10.0, connect=5.0)
headers = {"X-Client-Name": "MyCustomClient"}
# proxy_url = "http://user:password@myproxy.com:8080" # プロキシの例

client = httpx.Client(
    base_url="https://httpbin.org/",
    headers=headers,
    timeout=timeout,
    limits=limits,
    # proxies=proxy_url, # プロキシを使う場合
    follow_redirects=True,
    http2=False # HTTP/1.1 を強制する場合
)

try:
    with client: # with 文で使う
        response = client.get("/headers") # base_url からの相対パス
        response.raise_for_status()
        print("カスタムクライアントでのリクエスト成功:")
        # import json
        # print(json.dumps(response.json(), indent=2, ensure_ascii=False))
        # ヘッダーに 'X-Client-Name': 'MyCustomClient' が含まれているか確認
        received_headers = response.json().get('headers', {})
        print(f"X-Client-Name ヘッダー: {received_headers.get('X-Client-Name')}")

except httpx.RequestError as e:
    print(f"リクエストエラー: {e}")

イベントフック (Event Hooks)

リクエストの送信前やレスポンスの受信後などに特定の関数（フック）を呼び出すことができます。ロギング、モニタリング、リクエスト/レスポンスの改変などに利用できます。

event_hooks 引数に、イベント名 ('request' または 'response') と呼び出す関数のリストをマッピングした辞書を渡します。

import httpx
import time

def log_request(request):
    print(f"[Request Hook] > {request.method} {request.url}")
    # リクエストヘッダーにタイムスタンプを追加する例
    request.headers['X-Request-Timestamp'] = str(time.time())

def log_response(response):
    request = response.request # 対応するリクエストオブジェクトを取得
    elapsed = response.elapsed.total_seconds()
    print(f"[Response Hook] < {request.method} {request.url} - Status {response.status_code} ({elapsed:.2f}s)")
    # レスポンスヘッダーを確認する例
    # print(f"    Server Header: {response.headers.get('Server')}")

# イベントフックを設定してクライアントを作成
event_hooks = {'request': [log_request], 'response': [log_response]}

with httpx.Client(event_hooks=event_hooks) as client:
    print("イベントフック付きでリクエストを送信します...")
    client.get('https://httpbin.org/get')
    client.post('https://httpbin.org/post', data={'key': 'value'})
    print("リクエスト完了。")

Transport API と Mounting

httpx は低レベルの Transport API を提供しており、リクエストの送信方法を完全にカスタマイズできます。例えば、特定のドメインに対して異なる設定（プロキシ、SSL検証など）を持つトランスポートを「マウント」することが可能です。

これは非常に高度な機能であり、特定のユースケース（例: Microservices 環境で特定の内部サービスへの通信方法を変える）で役立ちます。

import httpx

# 例: example.com へのリクエストのみ特定のプロキシを使う Transport
custom_transport = httpx.HTTPTransport(proxy="http://specific-proxy.com:8080")
# 例: ローカルの WSGI/ASGI アプリケーション用 Transport
# app = MyWSGIApp()
# wsgi_transport = httpx.WSGITransport(app=app)

# Client に Transport をマウントする
# デフォルトの Transport は通常のインターネットアクセス用
# 特定のプレフィックスにマッチする場合、マウントした Transport が使われる
mounts = {
    "all://example.com": custom_transport,
    # "http://internal-service": wsgi_transport # WSGIアプリをマウントする例
}

with httpx.Client(mounts=mounts) as client:
    # このリクエストは custom_transport (プロキシ経由) で送信される
    response_example = client.get("https://example.com")
    print(f"example.com status: {response_example.status_code}")

    # このリクエストはデフォルトの Transport で送信される
    response_httpbin = client.get("https://httpbin.org/get")
    print(f"httpbin.org status: {response_httpbin.status_code}")

    # このリクエストは wsgi_transport で処理される (もしマウントされていれば)
    # response_internal = client.get("http://internal-service/status")
    # print(f"internal-service status: {response_internal.status_code}")

🎉 まとめ

httpx は、Python における HTTP 通信のための強力でモダンなライブラリです。requests の使いやすさを引き継ぎつつ、非同期処理と HTTP/2・HTTP/3 という現代的な Web 技術に対応している点が大きな魅力です。

特に、非同期 I/O が求められるアプリケーションや、最新のプロトコルを活用してパフォーマンスを向上させたい場合に、httpx は非常に有効な選択肢となります。タイムアウト、認証、プロキシ、イベントフック、トランスポート API など、豊富なカスタマイズオプションも提供されており、複雑な要件にも対応可能です。

もしあなたが Python で Web API を利用したり、Web スクレイピングを行ったり、あるいは非同期処理に関心があるなら、ぜひ httpx を試してみてください。きっとそのパワフルさと柔軟性に満足するはずです！💪

公式ドキュメント (https://www.python-httpx.org/) には、さらに詳細な情報や使用例が豊富に掲載されていますので、ぜひ参照してください。