PythonのhttpxライブラリでAPI通信を効率化！初心者向け入門ガイド

Python

最終更新2025年9月26日

リンドくん

たなべ先生、API通信でrequestsライブラリを使ってるんですけど、httpxっていうのもあるって聞きました。何が違うんですか？

たなべ

httpxはrequestsの次世代版と言えるライブラリなんだ。
特に非同期処理やHTTP/2対応など、現代のWeb開発に必要な機能が充実しているよ。

Pythonでウェブアプリを作るとき、必ず必要になるのがHTTP通信です。
天気予報を取得したり、SNSのAPIを使ったり、外部サービスとの連携は現代のプログラミングでは必須の技術ですよね。

これまでrequestsライブラリが広く使われてきましたが、最近注目されているのがhttpxという新しいライブラリです。
より高速で、より安全で、より使いやすいHTTP通信を実現できる優れたツールです。

この記事では、httpxの基本的な使い方から実践的な活用方法まで、初心者の方でも理解できるよう解説していきます。

プログラミング学習でお悩みの方へ

HackATAは、エンジニアを目指す方のためのプログラミング学習コーチングサービスです。経験豊富な現役エンジニアがあなたの学習をサポートします。

✓ 質問し放題

✓ β版公開中（2025年内の特別割引）

HackATAの詳細を見る

httpxとは？requestsとの違い

httpxの主な特徴

httpxは2019年に登場したPythonのHTTPクライアントライブラリで、以下の特徴があります。

requestsと互換性のあるAPI → 学習コストが低い
同期・非同期の両方に対応 → パフォーマンスが大幅向上
HTTP/2対応 → 最新のHTTP規格で高速通信
完全な型ヒント対応 → 開発効率が向上

なぜhttpxを選ぶべきか

requestsライブラリには以下の課題があります。

同期処理のみで、大量のリクエストに時間がかかる
HTTP/1.1のみ対応で、最新の高速化技術が使えない

httpxはこれらの問題を解決し、より高速で安全なHTTP通信を提供します。

httpxの基本的な使い方

インストール

pip install httpx
# uvの場合はuv add httpx

基本的なGETリクエスト

import httpx

# 基本的なGETリクエスト
response = httpx.get('https://api.github.com/users/octocat')

print(f"ステータスコード: {response.status_code}")
print(f"ユーザー名: {response.json()['name']}")

POSTリクエスト

import httpx

# JSONデータを送信
data = {
    "name": "田中太郎",
    "email": "tanaka@example.com"
}

response = httpx.post('https://httpbin.org/post', json=data)
print(f"送信結果: {response.status_code}")

ヘッダーとパラメータの設定

import httpx

headers = {'Authorization': 'Bearer your-token'}
params = {'page': 1, 'limit': 10}

response = httpx.get(
    'https://api.example.com/data',
    headers=headers,
    params=params
)

非同期処理でパフォーマンスを向上

リンドくん

非同期処理って具体的にどんなメリットがあるんですか？

たなべ

例えば3つのAPIを順番に呼ぶと3秒かかる処理が、非同期なら同時に呼んで1秒で完了できるんだ！

同期処理と非同期処理の比較

同期処理

import httpx
import time

def fetch_sync():
    start = time.time()
    urls = ['https://httpbin.org/delay/1'] * 3
    
    for url in urls:
        response = httpx.get(url)
        print(f"ステータス: {response.status_code}")
    
    print(f"実行時間: {time.time() - start:.2f}秒")  # 約3秒

fetch_sync()

非同期処理

import httpx
import asyncio
import time

async def fetch_async():
    start = time.time()
    urls = ['https://httpbin.org/delay/1'] * 3
    
    async with httpx.AsyncClient() as client:
        tasks = [client.get(url) for url in urls]
        responses = await asyncio.gather(*tasks)
        
        for response in responses:
            print(f"ステータス: {response.status_code}")
    
    print(f"実行時間: {time.time() - start:.2f}秒")  # 約1秒

asyncio.run(fetch_async())

エラーハンドリングとタイムアウト設定

基本的なエラーハンドリング

import httpx

def safe_request(url):
    try:
        response = httpx.get(url, timeout=10.0)
        response.raise_for_status()  # HTTPエラーの場合に例外発生
        return response.json()
        
    except httpx.RequestError:
        print("ネットワークエラーが発生しました")
        return None
        
    except httpx.HTTPStatusError as e:
        print(f"HTTPエラー {e.response.status_code}")
        return None

# 使用例
result = safe_request('https://api.example.com/data')
if result:
    print("データ取得成功")

詳細なタイムアウト設定

import httpx

# 詳細なタイムアウト設定
timeout = httpx.Timeout(
    connect=5.0,    # 接続タイムアウト
    read=10.0,      # 読み込みタイムアウト
    write=5.0,      # 書き込みタイムアウト
)

with httpx.Client(timeout=timeout) as client:
    response = client.get('https://api.example.com/data')

その他の活用例

APIクライアントクラスの作成

import httpx
from typing import Optional, Dict, Any

class APIClient:
    def __init__(self, base_url: str, api_key: Optional[str] = None):
        headers = {'Content-Type': 'application/json'}
        if api_key:
            headers['Authorization'] = f'Bearer {api_key}'
        
        self.client = httpx.Client(
            base_url=base_url,
            headers=headers,
            timeout=30.0
        )
    
    def get(self, endpoint: str) -> Dict[str, Any]:
        """データ取得"""
        response = self.client.get(endpoint)
        response.raise_for_status()
        return response.json()
    
    def post(self, endpoint: str, data: Dict[str, Any]) -> Dict[str, Any]:
        """データ作成"""
        response = self.client.post(endpoint, json=data)
        response.raise_for_status()
        return response.json()
    
    def close(self):
        self.client.close()

# 使用例
api = APIClient('https://jsonplaceholder.typicode.com')
try:
    user = api.get('/users/1')
    print(f"ユーザー名: {user['name']}")
finally:
    api.close()

非同期版のダッシュボード

import httpx
import asyncio

class AsyncDashboard:
    def __init__(self):
        self.client = httpx.AsyncClient(timeout=10.0)
    
    async def get_weather(self):
        """天気情報を取得（例）"""
        try:
            response = await self.client.get('https://httpbin.org/json')
            return {"service": "weather", "status": "success"}
        except:
            return {"service": "weather", "status": "error"}
    
    async def get_news(self):
        """ニュース情報を取得（例）"""
        try:
            response = await self.client.get('https://jsonplaceholder.typicode.com/posts/1')
            return {"service": "news", "status": "success"}
        except:
            return {"service": "news", "status": "error"}
    
    async def get_all_data(self):
        """すべてのデータを並列取得"""
        results = await asyncio.gather(
            self.get_weather(),
            self.get_news(),
            return_exceptions=True
        )
        return results
    
    async def close(self):
        await self.client.aclose()

# 使用例
async def main():
    dashboard = AsyncDashboard()
    try:
        data = await dashboard.get_all_data()
        for result in data:
            if isinstance(result, dict):
                print(f"{result['service']}: {result['status']}")
    finally:
        await dashboard.close()

asyncio.run(main())