Python httpxで爆速非同期HTTP通信を実装する

どうも、nippa です。

PythonでHTTP通信を行う際、従来のrequestsライブラリは同期処理のため大量のAPIリクエストで性能ボトルネックとなります。httpxライブラリを使用することで、非同期処理による高速なHTTP通信が実現できます。

環境

• Python 3.12
• httpx
• asyncio

httpxとrequestsの違い

requests（同期処理）

import requests
import time

def fetch_sync(urls):
    results = []
    start_time = time.time()

    for url in urls:
        response = requests.get(url)
        results.append(response.json())

    print(f"同期処理時間: {time.time() - start_time:.2f}秒")
    return results

httpx（非同期処理）

import httpx
import asyncio
import time

async def fetch_async(urls):
    results = []
    start_time = time.time()

    async with httpx.AsyncClient() as client:
        tasks = [client.get(url) for url in urls]
        responses = await asyncio.gather(*tasks)

        for response in responses:
            results.append(response.json())

    print(f"非同期処理時間: {time.time() - start_time:.2f}秒")
    return results

基本的な使用方法

インストール

pip install httpx

同期処理での使用

import httpx

# GETリクエスト
response = httpx.get('https://api.github.com/users/octocat')
print(response.json())

# POSTリクエスト
data = {'key': 'value'}
response = httpx.post('https://httpbin.org/post', json=data)
print(response.status_code)

# ヘッダー付きリクエスト
headers = {'Authorization': 'Bearer token'}
response = httpx.get('https://api.example.com/data', headers=headers)

非同期処理での使用

import httpx
import asyncio

async def main():
    async with httpx.AsyncClient() as client:
        # 単一リクエスト
        response = await client.get('https://api.github.com/users/octocat')
        print(response.json())

        # 複数の並行リクエスト
        urls = [
            'https://api.github.com/users/octocat',
            'https://api.github.com/users/defunkt',
            'https://api.github.com/users/pjhyett'
        ]

        tasks = [client.get(url) for url in urls]
        responses = await asyncio.gather(*tasks)

        for response in responses:
            print(f"Status: {response.status_code}, User: {response.json()['login']}")

# 実行
asyncio.run(main())

実践的な使用例

API レート制限を考慮した実装

import httpx
import asyncio
from asyncio import Semaphore

class RateLimitedClient:
    def __init__(self, rate_limit=10):
        self.semaphore = Semaphore(rate_limit)
        self.client = httpx.AsyncClient(
            timeout=httpx.Timeout(10.0),
            limits=httpx.Limits(max_keepalive_connections=20, max_connections=100)
        )

    async def get(self, url):
        async with self.semaphore:
            try:
                response = await self.client.get(url)
                response.raise_for_status()
                return response
            except httpx.HTTPError as e:
                print(f"HTTP error occurred: {e}")
                return None

    async def close(self):
        await self.client.aclose()

async def fetch_with_rate_limit():
    client = RateLimitedClient(rate_limit=5)

    urls = [f'https://jsonplaceholder.typicode.com/posts/{i}' for i in range(1, 21)]

    tasks = [client.get(url) for url in urls]
    responses = await asyncio.gather(*tasks, return_exceptions=True)

    valid_responses = [r for r in responses if r and not isinstance(r, Exception)]
    print(f"成功: {len(valid_responses)}/{len(urls)} リクエスト")

    await client.close()

asyncio.run(fetch_with_rate_limit())

エラーハンドリングとリトライ機能

import httpx
import asyncio
from typing import List, Optional

async def fetch_with_retry(
    client: httpx.AsyncClient,
    url: str,
    max_retries: int = 3
) -> Optional[httpx.Response]:

    for attempt in range(max_retries):
        try:
            response = await client.get(url)
            response.raise_for_status()
            return response

        except httpx.TimeoutException:
            print(f"Timeout for {url}, attempt {attempt + 1}")
            if attempt == max_retries - 1:
                return None
            await asyncio.sleep(2 ** attempt)  # 指数バックオフ

        except httpx.HTTPStatusError as e:
            if e.response.status_code >= 500:
                print(f"Server error {e.response.status_code} for {url}")
                if attempt == max_retries - 1:
                    return None
                await asyncio.sleep(2 ** attempt)
            else:
                print(f"Client error {e.response.status_code} for {url}")
                return None

        except Exception as e:
            print(f"Unexpected error for {url}: {e}")
            return None

    return None

async def batch_fetch_with_retry():
    urls = [
        'https://jsonplaceholder.typicode.com/posts/1',
        'https://jsonplaceholder.typicode.com/posts/2',
        'https://httpbin.org/delay/2',  # 遅いエンドポイント
        'https://httpbin.org/status/500',  # エラーエンドポイント
    ]

    async with httpx.AsyncClient(timeout=5.0) as client:
        tasks = [fetch_with_retry(client, url) for url in urls]
        responses = await asyncio.gather(*tasks)

        successful = [r for r in responses if r is not None]
        print(f"成功したリクエスト: {len(successful)}/{len(urls)}")

        for response in successful:
            print(f"Status: {response.status_code}, URL: {response.url}")

asyncio.run(batch_fetch_with_retry())

ストリーミング処理

import httpx
import asyncio

async def download_large_file():
    url = 'https://httpbin.org/stream/1000'

    async with httpx.AsyncClient() as client:
        async with client.stream('GET', url) as response:
            print(f"Content-Length: {response.headers.get('content-length')}")

            total_size = 0
            async for chunk in response.aiter_bytes(chunk_size=8192):
                total_size += len(chunk)
                # ここでチャンクを処理
                print(f"Downloaded: {total_size} bytes", end='\r')

            print(f"\nTotal downloaded: {total_size} bytes")

asyncio.run(download_large_file())

パフォーマンス比較

100個のAPIリクエストの処理時間

import time
import asyncio
import httpx
import requests

def benchmark_sync():
    urls = [f'https://jsonplaceholder.typicode.com/posts/{i}'
            for i in range(1, 101)]

    start_time = time.time()
    for url in urls:
        response = requests.get(url)
    sync_time = time.time() - start_time
    return sync_time

async def benchmark_async():
    urls = [f'https://jsonplaceholder.typicode.com/posts/{i}'
            for i in range(1, 101)]

    start_time = time.time()
    async with httpx.AsyncClient() as client:
        tasks = [client.get(url) for url in urls]
        await asyncio.gather(*tasks)
    async_time = time.time() - start_time
    return async_time

# ベンチマーク実行
sync_time = benchmark_sync()
async_time = asyncio.run(benchmark_async())

print(f"同期処理: {sync_time:.2f}秒")
print(f"非同期処理: {async_time:.2f}秒")
print(f"速度向上: {sync_time/async_time:.1f}倍")

ベストプラクティス

1. AsyncClientの適切な管理

# 推奨: コンテキストマネージャーを使用
async with httpx.AsyncClient() as client:
    response = await client.get(url)

# 非推奨: 手動でのクローズ管理
client = httpx.AsyncClient()
try:
    response = await client.get(url)
finally:
    await client.aclose()

2. 適切なタイムアウト設定

timeout = httpx.Timeout(
    connect=5.0,    # 接続タイムアウト
    read=10.0,      # 読み取りタイムアウト
    write=5.0,      # 書き込みタイムアウト
    pool=10.0       # プールタイムアウト
)

async with httpx.AsyncClient(timeout=timeout) as client:
    response = await client.get(url)

3. コネクションプールの最適化

limits = httpx.Limits(
    max_keepalive_connections=20,
    max_connections=100,
    keepalive_expiry=30.0
)

async with httpx.AsyncClient(limits=limits) as client:
    # リクエスト処理
    pass

感想

httpxライブラリを使った非同期HTTP処理により、従来のrequestsと比べて大幅な性能向上が期待できます。特に複数のAPIを並行して処理する場面では、その効果は絶大です。適切なエラーハンドリングとレート制限を組み合わせることで、実用的な高性能HTTPクライアントを構築できます。

ではでは、また次回。