次世代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/) には、さらに詳細な情報や使用例が豊富に掲載されていますので、ぜひ参照してください。