urllib3

Pythonの低レベルHTTPクライアントライブラリ。多くのHTTPクライアント(Requestsを含む)の基盤として機能。接続プーリング、SSL/TLS、圧縮、ストリーミング、詳細な制御機能を提供。カスタマイズ性に優れ、月1.5億回以上のダウンロード数を記録。

HTTPクライアントPython低レベルコネクションプールスレッドセーフ

GitHub概要

urllib3/urllib3

urllib3 is a user-friendly HTTP client library for Python

スター3,905
ウォッチ99
フォーク1,194
作成日:2011年9月18日
言語:Python
ライセンス:MIT License

トピックス

httphttp-clientpythonurllib3

スター履歴

urllib3/urllib3 Star History
データ取得日時: 2025/7/18 01:39

ライブラリ

urllib3

概要

urllib3は、Pythonの低レベルHTTPクライアントライブラリです。多くの高レベルライブラリ(requests等)の基盤として使用され、コネクションプーリング、SSL/TLS検証、スレッドセーフ性を提供します。2025年においても、詳細なHTTP通信制御が必要な場面で重要な役割を果たしています。

詳細

urllib3は「urllib」の改良版として開発され、標準ライブラリの制限を克服した高機能HTTPクライアントです。requestsライブラリの基盤として採用されており、Python HTTPエコシステムの根幹を支えています。WebAssembly対応やHTTP/2サポートなど、現代的なWeb技術への対応も進んでいます。

主な特徴

  • コネクションプーリング: 効率的な接続再利用
  • スレッドセーフ: マルチスレッド環境での安全な使用
  • SSL/TLS検証: 包括的なセキュリティ機能
  • リトライ機能: 自動的な接続再試行
  • プロキシサポート: HTTP/HTTPS/SOCKSプロキシ対応
  • 圧縮対応: gzip、deflate、brオートデコード
  • カスタムヘッダー: 詳細なリクエスト制御
  • アップロード支援: マルチパートフォームデータ
  • タイムアウト制御: 詳細なタイムアウト設定
  • WebAssembly対応: モダン環境での実行支援

メリット・デメリット

メリット

  • 高パフォーマンスと効率的なリソース利用
  • requestsをはじめ多くのライブラリの基盤として実績豊富
  • 詳細なHTTP制御とカスタマイゼーションが可能
  • スレッドセーフでマルチスレッド環境に対応
  • 豊富なセキュリティ機能とSSL/TLS対応

デメリット

  • 高レベルライブラリと比較して学習コストが高い
  • 冗長なコード記述が必要
  • 非同期処理(async/await)は未対応
  • JSON処理やセッション管理は手動実装が必要
  • エラーハンドリングが低レベルで複雑

参考ページ

書き方の例

基本的なセットアップ

import urllib3

# HTTP接続プールの作成
http = urllib3.PoolManager()

# SSL証明書検証を無効にする場合(開発時のみ)
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
http_insecure = urllib3.PoolManager(cert_reqs='CERT_NONE')

基本的なリクエスト

import urllib3
import json

http = urllib3.PoolManager()

# GETリクエスト
response = http.request('GET', 'https://httpbin.org/get')
print(f"Status: {response.status}")
print(f"Data: {response.data.decode('utf-8')}")

# POSTリクエスト(JSON)
data = {"name": "田中太郎", "age": 30}
encoded_data = json.dumps(data).encode('utf-8')

response = http.request(
    'POST',
    'https://httpbin.org/post',
    body=encoded_data,
    headers={'Content-Type': 'application/json'}
)

# ヘッダー付きリクエスト
headers = {
    'User-Agent': 'MyApp/1.0',
    'Accept': 'application/json',
    'Authorization': 'Bearer your-token-here'
}

response = http.request(
    'GET',
    'https://api.example.com/data',
    headers=headers
)

セッション管理(コネクションプーリング)

import urllib3

# カスタマイズされたプールマネージャー
http = urllib3.PoolManager(
    num_pools=50,           # プール数
    maxsize=100,            # プールあたりの最大接続数
    block=True,             # 接続制限時にブロック
    timeout=urllib3.Timeout(connect=2.0, read=10.0)
)

# 特定ホスト用の接続プール
host_pool = urllib3.HTTPSConnectionPool(
    'api.example.com',
    port=443,
    maxsize=20,
    timeout=10.0
)

# 接続プールを使用したリクエスト
response = host_pool.request('GET', '/api/v1/users')
print(f"Connection pool info: {host_pool.pool}")

認証とSSL設定

import urllib3
from urllib3.util import make_headers

# Basic認証
auth_headers = make_headers(basic_auth='username:password')
http = urllib3.PoolManager()

response = http.request(
    'GET',
    'https://httpbin.org/basic-auth/username/password',
    headers=auth_headers
)

# クライアント証明書を使用したSSL認証
http_with_cert = urllib3.PoolManager(
    cert_file='/path/to/client.crt',
    key_file='/path/to/client.key',
    ca_certs='/path/to/ca.crt'
)

response = http_with_cert.request('GET', 'https://secure-api.example.com')

# カスタムSSL設定
import ssl

ssl_context = ssl.create_default_context()
ssl_context.check_hostname = False
ssl_context.verify_mode = ssl.CERT_NONE

http_custom_ssl = urllib3.PoolManager(
    ssl_context=ssl_context
)

エラーハンドリング

import urllib3
from urllib3.exceptions import (
    HTTPError, 
    ConnectTimeoutError, 
    ReadTimeoutError,
    SSLError,
    MaxRetryError
)

http = urllib3.PoolManager()

try:
    response = http.request(
        'GET', 
        'https://example.com/api/data',
        timeout=urllib3.Timeout(connect=5, read=10),
        retries=urllib3.Retry(
            total=3,
            backoff_factor=0.3,
            status_forcelist=[500, 502, 503, 504]
        )
    )
    
    if response.status == 200:
        data = response.data.decode('utf-8')
        print(f"Success: {data}")
    else:
        print(f"HTTP Error: {response.status}")
        
except ConnectTimeoutError:
    print("接続タイムアウトエラー")
except ReadTimeoutError:
    print("読み込みタイムアウトエラー")
except SSLError as e:
    print(f"SSL/TLS エラー: {e}")
except MaxRetryError as e:
    print(f"最大リトライ回数に達しました: {e}")
except HTTPError as e:
    print(f"HTTP エラー: {e}")

パフォーマンス最適化

import urllib3
import threading
from concurrent.futures import ThreadPoolExecutor

# ストリーミングダウンロード
http = urllib3.PoolManager()

response = http.request(
    'GET',
    'https://example.com/large-file.zip',
    preload_content=False  # ストリーミングモード
)

# チャンク単位での読み込み
with open('downloaded_file.zip', 'wb') as f:
    for chunk in response.stream(1024):  # 1KBずつ読み込み
        f.write(chunk)

response.release_conn()  # 接続を明示的にリリース

# 圧縮対応
response = http.request(
    'GET',
    'https://api.example.com/data',
    headers={'Accept-Encoding': 'gzip, deflate, br'}
)

# 並行リクエスト処理
def fetch_url(url):
    try:
        response = http.request('GET', url)
        return {'url': url, 'status': response.status, 'length': len(response.data)}
    except Exception as e:
        return {'url': url, 'error': str(e)}

urls = [
    'https://httpbin.org/delay/1',
    'https://httpbin.org/delay/2',
    'https://httpbin.org/delay/3'
]

# ThreadPoolExecutorを使用した並行処理
with ThreadPoolExecutor(max_workers=5) as executor:
    results = list(executor.map(fetch_url, urls))
    
for result in results:
    print(result)