Python Requests ライブラリスニペット

1 Python Requestsの紹介

requestsライブラリは、Pythonの洗練されたHTTPクライアントであり、API呼び出しを自然なものにします。HTTPの煩雑な詳細を抽象化しつつ、その強力な機能を内部に保持しています。インターネットのための信頼できる宅配便業者と考えてください。簡単なページ取得から複雑な認証フローまで、すべてを扱います。

開発者に愛される理由:

人間が理解しやすい: APIは、暗号的なHTTP専門用語ではなく、平易な英語のように読み取れます。
組み込みJSON: 自動解析により、定型的な解析の苦労がなくなります。
セッションサポート: クッキーとコネクションプーリングを自動的に保持します。
堅牢なエラー処理: 暗号的なステータスコードの代わりに、明確な例外を発生させます。
SSL検証: デフォルトで安全、必要に応じて柔軟に対応します。

データを取得する準備はできましたか？基本から始めましょう。

2 前提条件：Python & Pip

requestsライブラリをインストールするには、Python 3.6+とpipが必要です。セットアップを確認しましょう。

python --version && pip --version

これらのコマンドが機能すれば、問題ありません。そうでなければ、python.orgからPythonをダウンロードしてください。pipも一緒にインストールされます。

3 Requestsライブラリのインストール

まず、新しい仮想環境を作成しましょう。これにより、プロジェクトの依存関係がクリーンに保たれ、バージョン競合を防ぐことができます。

python -m venv .venv && source .venv/bin/activate  # macOS/Linux
# または
python -m venv .venv && .venv\Scripts\activate  # Windows

次に、pipを使用してrequestsをインストールします。

pip install requests

4 最初のGETリクエスト

最も一般的な操作であるデータ取得から始めましょう。この簡単なスニペットは、公開APIからJSONレスポンスを取得します。

import requests

response = requests.get('https://httpbin.org/json')
print(f"ステータス: {response.status_code}")
print(f"Content-Type: {response.headers['content-type']}")

# 自動JSON解析
data = response.json()
print(f"キー: {list(data.keys())}")

自動JSON解析に注目してください。requestsはContent-Typeを検出し、自動的に解析します。手動でのjson.loads()の処理は不要です。

5 データを含むPOSTリクエスト

サーバーにデータを送信しますか？ POSTリクエストは、フォームデータ、JSON、ファイルをシームレスに処理します。一般的なパターンを以下に示します。

import requests

# フォームデータを送信
form_data = {'username': 'alice', 'password': 'secret123'}
response = requests.post('https://httpbin.org/post', data=form_data)

# JSONデータを送信 (Content-Typeヘッダーを自動設定)
json_data = {'name': 'Alice', 'email': '[email protected]'}
response = requests.post('https://httpbin.org/post', json=json_data)

print(f"ステータス: {response.status_code}")
print(f"レスポンス: {response.json()}")

jsonパラメータは、適切なヘッダーを自動的に設定し、データをシリアライズします。手動でヘッダーを設定するよりもはるかにクリーンです。

6 カスタムヘッダーと認証

APIはカスタムヘッダーや認証を必要とすることがよくあります。requestsは、組み込みの認証クラスでこれを簡単に実現します。

import requests
from requests.auth import HTTPBasicAuth, HTTPDigestAuth

# カスタムヘッダー
headers = {
    'User-Agent': 'MyApp/1.0',
    'Accept': 'application/json',
    'X-API-Key': 'your-api-key-here'
}

response = requests.get('https://httpbin.org/headers', headers=headers)
print(response.json())

# Basic認証
auth = HTTPBasicAuth('username', 'password')
response = requests.get('https://httpbin.org/basic-auth/user/pass', auth=auth)

# ヘッダーにAPIキー
api_key = 'sk-1234567890abcdef'
headers = {'Authorization': f'Bearer {api_key}'}
response = requests.get('https://httpbin.org/bearer', headers=headers)

HTTPBasicAuthとHTTPDigestAuthは、エンコーディングを自動的に処理します。最新のAPIでは、AuthorizationヘッダーのBearerトークンが標準です。

7 エラー処理とステータスコード

堅牢なアプリケーションは、エラーを適切に処理します。requestsはネットワーク問題に対して例外を発生させ、詳細なレスポンスオブジェクトを提供します。

import requests
from requests.exceptions import RequestException, Timeout, ConnectionError

try:
    response = requests.get('https://httpbin.org/status/404', timeout=5)
    
    # 不正なステータスコードに対して例外を発生させる
    response.raise_for_status()
    
except Timeout:
    print("タイムアウト（5秒後）")
except ConnectionError:
    print("サーバーへの接続に失敗しました")
except RequestException as e:
    print(f"リクエスト失敗: {e}")
    
else:
    print(f"成功！ ステータス: {response.status_code}")
    print(f"ヘッダー: {response.headers}")
    print(f"ボディ: {response.text[:100]}...")  # 最初の100文字
    
# 手動でのステータスコードチェック
if response.status_code == 200:
    data = response.json()
    print("データ受信:", data)
elif response.status_code == 404:
    print("リソースが見つかりません")
else:
    print(f"予期しないステータス: {response.status_code}")

raise_for_status()は、4xx/5xxレスポンスに対してHTTPErrorを発生させます。タイムアウトは、アプリケーションが無限にハングするのを防ぎます。

8 JSON処理とデータハンドリング

JSONデータを効率的に扱うことが、requestsの真価です。エンコーディング、デコーディング、検証を自動的に処理します。

import requests

# 複雑なJSON処理
response = requests.get('https://httpbin.org/json')

# エラー処理を含む安全なJSON解析
try:
    data = response.json()
    print(f"タイプ: {type(data)}")
    print(f"キー: {data.keys() if isinstance(data, dict) else '辞書ではありません'}")
except ValueError as e:
    print(f"JSONが無効です: {e}")
    print(f"生のコンテンツ: {response.text}")

# ネストされたデータとの作業
user_data = {
    'name': 'Alice Johnson',
    'email': '[email protected]',
    'preferences': {
        'theme': 'dark',
        'notifications': True,
        'language': 'en'
    }
}

# 複雑なネストされたJSONを送信
response = requests.post('https://httpbin.org/post', json=user_data)
result = response.json()

# ネストされた値へのアクセス
nested = result.get('json', {}).get('preferences', {})
theme = nested.get('theme', 'light')
print(f"ユーザーのテーマ: {theme}")

常にget()のような安全なメソッドを使用してネストされたJSONにアクセスし、KeyError例外を回避してください。

9 ファイルダウンロードとストリーミング

ファイルを効率的にダウンロードすることで、メモリ問題を回避できます。大きなファイルにはストリーミングを使用し、Content-Typeを確認してください。

import requests
import os
from pathlib import Path

# ファイルをダウンロード
url = 'https://httpbin.org/image/jpeg'
response = requests.get(url, stream=True)

if response.status_code == 200:
    # downloadsディレクトリを作成
    downloads_dir = Path('downloads')
    downloads_dir.mkdir(exist_ok=True)
    
    # ファイルを保存
    file_path = downloads_dir / 'image.jpg'
    with open(file_path, 'wb') as f:
        for chunk in response.iter_content(chunk_size=8192):
            if chunk:  # keep-aliveチャンクをフィルタリング
                f.write(chunk)
    
    print(f"ダウンロード完了: {file_path} ({file_path.stat().st_size} バイト)")
else:
    print(f"ダウンロード失敗: {response.status_code}")

# ダウンロード前にファイル情報を取得
head_response = requests.head(url)
content_length = head_response.headers.get('content-length')
content_type = head_response.headers.get('content-type')

print(f"ファイルサイズ: {content_length} バイト")
print(f"Content-Type: {content_type}")

# 進捗表示付きでダウンロード
def download_with_progress(url, filename):
    response = requests.get(url, stream=True)
    total_size = int(response.headers.get('content-length', 0))
    
    with open(filename, 'wb') as f:
        downloaded = 0
        for chunk in response.iter_content(chunk_size=1024):
            if chunk:
                f.write(chunk)
                downloaded += len(chunk)
                
                # 進捗を表示
                if total_size > 0:
                    progress = (downloaded / total_size) * 100
                    print(f"進捗: {progress:.1f}%")

ストリーミングダウンロードは、ファイル全体をメモリにロードすることを防ぎます。大きな画像、ドキュメント、メディアファイルに最適です。

10 セッション管理とクッキー

ログインセッションなど、リクエスト間で状態を維持する必要があるアプリケーションには、requestsのSessionオブジェクトを使用します。

import requests

# 持続的な接続とクッキーのためのセッションを作成
session = requests.Session()

# すべてのリクエストのデフォルトヘッダーを設定
session.headers.update({
    'User-Agent': 'MyApp/1.0',
    'Accept': 'application/json'
})

# クッキー、コネクションプーリング、リダイレクトを自動処理
response1 = session.get('https://httpbin.org/cookies/set/session_id/abc123')
print("クッキー設定:", response1.cookies.get_dict())

# 後続のリクエストはクッキーを含みます
response2 = session.get('https://httpbin.org/cookies')
print("受信クッキー:", response2.json())

# セッションレベルの設定
session.auth = ('username', 'password')
session.timeout = 10

# コンテキストマネージャーを使用して自動クリーンアップ
with requests.Session() as s:
    response = s.get('https://httpbin.org/json')
    data = response.json()
    print("コンテキストマネージャーを使用")
    
# 手動クッキー処理
cookies = {'sessionid': 'abc123', 'csrftoken': 'xyz789'}
response = requests.get('https://httpbin.org/cookies', cookies=cookies)
print("手動クッキー:", response.json())

セッションはコネクションプーリングを自動的に処理するため、後続のリクエストが高速になり、サーバー負荷が軽減されます。

11 ベストプラクティスと一般的なパターン

プロフェッショナルなHTTPクライアントは、特定のパターンに従います。ここでは、本番コードのための実績のあるプラクティスを紹介します。

import requests
import logging
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

# デバッグ用のロギングを設定
logging.basicConfig(level=logging.WARNING)  # ノイズを減らす
logger = logging.getLogger(__name__)

class APIClient:
    def __init__(self, base_url, timeout=10, max_retries=3):
        self.session = requests.Session()
        
        # 一時的な障害に対するリトライ戦略
        retry_strategy = Retry(
            total=max_retries,
            status_forcelist=[429, 500, 502, 503, 504],
            method_whitelist=["HEAD", "GET", "OPTIONS"],
            backoff_factor=1
        )
        
        adapter = HTTPAdapter(max_retries=retry_strategy)
        self.session.mount("http://", adapter)
        self.session.mount("https://", adapter)
        
        self.base_url = base_url
        self.timeout = timeout
    
    def get(self, endpoint, **kwargs):
        """エラー処理付きGETリクエストを実行"""
        url = f"{self.base_url.rstrip('/')}/{endpoint.lstrip('/')}"
        
        try:
            response = self.session.get(url, timeout=self.timeout, **kwargs)
            response.raise_for_status()
            return response
        except requests.RequestException as e:
            logger.error(f"リクエスト失敗: {e}")
            raise
    
    def post(self, endpoint, **kwargs):
        """POSTリクエストを実行"""
        url = f"{self.base_url.rstrip('/')}/{endpoint.lstrip('/')}"
        return self.session.post(url, timeout=self.timeout, **kwargs)

# 使用例
api = APIClient('https://httpbin.org')
try:
    response = api.get('json')
    data = response.json()
    print(f"データ取得: {list(data.keys())}")
except Exception as e:
    print(f"API呼び出し失敗: {e}")

# 環境ベースの設定
import os

API_BASE_URL = os.getenv('API_BASE_URL', 'https://httpbin.org')
API_KEY = os.getenv('API_KEY', 'dev-key')

headers = {
    'Authorization': f'Bearer {API_KEY}',
    'User-Agent': f'MyApp/{__version__}'
}

これらのパターンは、リトライ、適切なエラーログ記録、環境設定を処理し、APIインタラクションのためのクリーンな抽象化レイヤーを作成します。

主なポイント

• コネクションプーリングとクッキー管理にはセッションを使用する
• 常にタイムアウトとエラー処理を実装する
• 自動シリアライズにはjson=パラメータを活用する
• メモリ管理のために大きなファイルのダウンロードはストリーミングする
• 環境変数を使用して設定を行う

Python Requests ライブラリ スニペット

チュートリアルの進捗