Python Requests チートシート

HTTPリクエストを送信するための基本的な関数群です。

GET リクエスト

指定したURLからリソースを取得します。

import requests

# シンプルなGETリクエスト
response = requests.get('https://httpbin.org/get')

# レスポンスのステータスコード確認
print(f"ステータスコード: {response.status_code} {response.reason}")

# レスポンスの内容 (テキスト) を表示
# print(response.text)

POST リクエスト (フォームデータ)

サーバーにデータを送信します。data引数に辞書を渡すと、フォームエンコードされたデータとして送信されます。

import requests

payload = {'key1': 'value1', 'key2': 'value2'}

# data引数に辞書を渡してPOSTリクエスト
response = requests.post('https://httpbin.org/post', data=payload)

# レスポンスの内容 (JSON) を表示
print(response.json())

POST リクエスト (JSONデータ)

サーバーにJSONデータを送信します。json引数にPythonの辞書やリストを渡すと、自動的にJSON文字列にシリアライズされ、Content-Type: application/json ヘッダーが付与されます。

import requests

payload = {'some': 'data', 'nested': {'value': 123}}

# json引数に辞書を渡してPOSTリクエスト
response = requests.post('https://httpbin.org/post', json=payload)

# レスポンスの内容 (JSON) を表示
print(response.json())

その他のHTTPメソッド

PUT, DELETE, HEAD, OPTIONS など、他のHTTPメソッドも同様の形式で利用できます。

import requests

# PUTリクエスト (データ更新)
response_put = requests.put('https://httpbin.org/put', data={'key': 'value'})
print(f"PUT Status: {response_put.status_code}")

# DELETEリクエスト (リソース削除)
response_delete = requests.delete('https://httpbin.org/delete')
print(f"DELETE Status: {response_delete.status_code}")

# HEADリクエスト (ヘッダーのみ取得)
response_head = requests.head('https://httpbin.org/get')
print(f"HEAD Headers: {response_head.headers}")

# OPTIONSリクエスト (利用可能なメソッド確認)
response_options = requests.options('https://httpbin.org/get')
print(f"OPTIONS Allowed Methods: {response_options.headers.get('Allow')}")

サーバーからのレスポンスを扱うための属性やメソッドです。

try:
    response.raise_for_status()
    print("リクエスト成功！")
except requests.exceptions.HTTPError as err:
    print(f"HTTPエラーが発生しました: {err}")

リクエストに追加情報を付与する方法です。

URLパラメータ (クエリ文字列) の送信

params引数に辞書を渡すと、URLの末尾に ?key1=value1&key2=value2 のような形式で付加されます。

import requests

# パラメータを辞書で定義
payload = {'key1': 'value1', 'key2': ['value2a', 'value2b']} # 同じキーで複数の値も可

# params引数でGETリクエスト
response = requests.get('https://httpbin.org/get', params=payload)

# 送信されたURLを確認
print(response.url)
# 出力例: https://httpbin.org/get?key1=value1&key2=value2a&key2=value2b

カスタムヘッダーの送信

headers引数に辞書を渡すことで、任意のHTTPヘッダーを送信できます。User-Agentの偽装やAPIキーの送信などによく使われます。

import requests

url = 'https://httpbin.org/headers'
headers = {
    'User-Agent': 'My Awesome Python App/1.0',
    'Accept': 'application/json',
    'X-Api-Key': 'your_secret_api_key_here' # 例: APIキー
}

response = requests.get(url, headers=headers)

# 送信されたヘッダーを確認 (レスポンスの 'headers' フィールドに含まれる)
print(response.json()['headers'])

files引数を使用して、マルチパートフォームエンコード (multipart/form-data) でファイルをアップロードします。

単一ファイルのアップロード

辞書のキーがフォームのフィールド名、値がファイルオブジェクトになります。

import requests

url = 'https://httpbin.org/post'
# ファイルを開く (バイナリモード 'rb' が推奨)
try:
    with open('my_document.txt', 'rb') as f:
        files = {'file_field_name': f} # 'file_field_name' はサーバー側で期待されるフィールド名
        response = requests.post(url, files=files)
        print(response.json())
except FileNotFoundError:
    print("エラー: my_document.txt が見つかりません。")
except requests.exceptions.RequestException as e:
    print(f"リクエストエラー: {e}")

# ファイルオブジェクトを直接渡す代わりに、タプルで詳細を指定することも可能
# files = {'file': ('report.xls', open('report.xls', 'rb'), 'application/vnd.ms-excel', {'Expires': '0'})}
# ('ファイル名', ファイルオブジェクト, Content-Type, カスタムヘッダー)

ファイルは通常バイナリモード ('rb') で開きます。テキストファイルをテキストモード ('rt') で開いてアップロードすることも可能ですが、エンコーディングの問題が発生する可能性があるため、バイナリモードが安全です。Requestsはファイルの内容に基づいて Content-Type を推測しようとしますが、明示的に指定することも可能です（後述）。

複数ファイルのアップロード

files引数に複数のキーと値のペアを持つ辞書を渡します。

import requests

url = 'https://httpbin.org/post'
try:
    files_to_upload = {
        'text_file': open('document.txt', 'rb'),
        'image_file': open('logo.png', 'rb')
    }
    response = requests.post(url, files=files_to_upload)
    print(response.json()['files']) # アップロードされたファイルの情報
except FileNotFoundError:
    print("エラー: 指定されたファイルが見つかりません。")
except requests.exceptions.RequestException as e:
    print(f"リクエストエラー: {e}")
finally:
    # ファイルハンドルを閉じる (with文を使わない場合)
    if 'files_to_upload' in locals():
        for f in files_to_upload.values():
            if hasattr(f, 'close'):
                f.close()

ファイル名、Content-Type、カスタムヘッダーの指定

files辞書の値として、ファイルオブジェクトの代わりにタプルを使うことで、ファイル名、Content-Type、追加のヘッダーを明示的に指定できます。

import requests

url = 'https://httpbin.org/post'
try:
    with open('data.csv', 'rb') as f:
        files = {
            'csv_data': ('report.csv', f, 'text/csv', {'X-Custom-Header': 'hello'})
            # タプル形式: (ファイル名, ファイルオブジェクト, Content-Type, ヘッダー辞書)
        }
        response = requests.post(url, files=files)
        print(response.json())
except FileNotFoundError:
    print("エラー: data.csv が見つかりません。")
except requests.exceptions.RequestException as e:
    print(f"リクエストエラー: {e}")

# ファイルオブジェクトの代わりにコンテンツを直接指定することも可能
files = {'file': ('dummy.txt', b'This is the file content.', 'text/plain')}
response_direct = requests.post(url, files=files)
print(response_direct.json())

リクエストに認証情報を付加する方法です。

Basic認証

auth引数にユーザー名とパスワードのタプルを渡します。

import requests
from requests.auth import HTTPBasicAuth

url = 'https://httpbin.org/basic-auth/user/pass' # 例: 'user'/'pass' が認証情報

# 方法1: タプルを直接渡す
response1 = requests.get(url, auth=('user', 'pass'))
print(f"方法1 ステータス: {response1.status_code}") # 認証成功なら 200

# 方法2: HTTPBasicAuthオブジェクトを使用
auth = HTTPBasicAuth('user', 'pass')
response2 = requests.get(url, auth=auth)
print(f"方法2 ステータス: {response2.status_code}")

# 認証失敗の例 (間違ったパスワード)
try:
    response_fail = requests.get(url, auth=('user', 'wrongpass'))
    response_fail.raise_for_status() # 401 Unauthorized エラーが発生するはず
except requests.exceptions.HTTPError as e:
    print(f"認証失敗: {e.response.status_code} {e.response.reason}")

Digest認証

requests.auth.HTTPDigestAuth を使用します。

import requests
from requests.auth import HTTPDigestAuth

url = 'https://httpbin.org/digest-auth/auth/user/pass' # 例: Digest認証のエンドポイント
auth = HTTPDigestAuth('user', 'pass')

response = requests.get(url, auth=auth)
print(f"Digest認証 ステータス: {response.status_code}") # 認証成功なら 200

Digest認証はBasic認証より安全ですが、サーバーとクライアントの両方が対応している必要があります。

カスタム認証 (トークンなど)

多くの場合、APIキーやOAuthトークンはカスタムヘッダー (例: Authorization: Bearer YOUR_TOKEN) を使って送信します。これは前述の「カスタムヘッダーの送信」で対応できます。

より複雑な認証メカニズム（例: 署名生成）の場合は、requests.auth.AuthBase を継承してカスタム認証クラスを作成することも可能です。

import requests

# 例: Bearerトークンをヘッダーで送信
token = 'your_super_secret_token'
headers = {'Authorization': f'Bearer {token}'}
response = requests.get('https://api.example.com/protected/resource', headers=headers)
print(response.status_code)

# --- カスタム認証クラスの例 (概念) ---
from requests.auth import AuthBase

class TokenAuth(AuthBase):
    """カスタムトークン認証を付与する"""
    def __init__(self, token):
        self.token = token

    def __call__(self, r):
        # リクエストオブジェクトrを修正する
        r.headers['X-TokenAuth'] = f'{self.token}'
        return r

# 使用例
# response = requests.get('https://httpbin.org/get', auth=TokenAuth('my-secret-token'))
# print(response.request.headers)

複数のリクエスト間でクッキーや接続設定を持続させるためにセッションオブジェクトを使用します。

セッションオブジェクトの作成と利用

requests.Session() でセッションオブジェクトを作成し、そのオブジェクトのメソッド (get, post など) を使ってリクエストを送信します。これにより、サーバーからセットされたクッキーはセッションオブジェクト内に保持され、後続のリクエストで自動的に送信されます。

import requests

# セッションオブジェクトを作成
s = requests.Session()

# セッション経由で最初のGETリクエスト (クッキーがセットされることを期待)
s.get('https://httpbin.org/cookies/set/sessioncookie/123456789')

# 2回目のリクエスト (セッションクッキーが自動で送信される)
response = s.get('https://httpbin.org/cookies')

# レスポンスで送信されたクッキーを確認
print("セッションで送信されたクッキー:")
print(response.json()) # {'cookies': {'sessioncookie': '123456789'}} が含まれるはず

# セッションオブジェクトにデフォルトヘッダーを設定することも可能
s.headers.update({'x-test-header': 'true'})
response_with_header = s.get('https://httpbin.org/headers')
print("\nセッションのデフォルトヘッダー:")
print(response_with_header.json()['headers'])

# セッションオブジェクトにデフォルトパラメータを設定
# s.params.update({'param1': 'value1'})

# セッションオブジェクトにデフォルト認証情報を設定
# s.auth = ('user', 'pass')

# セッションを使い終わったら閉じるのが良い場合もある (TCP接続の再利用など)
s.close()

同じホストに対して複数回リクエストを行う場合、セッションオブジェクトを使用すると、TCP接続が再利用されることがあり、パフォーマンスが向上します。また、ログイン状態の維持など、クッキー管理が重要な場合に非常に便利です。

タイムアウト、リダイレクト、SSL検証、プロキシなど、リクエストの挙動を細かく制御します。

タイムアウト設定 ⏱️

timeout引数で、リクエストのタイムアウト秒数を指定します。接続タイムアウトと読み取りタイムアウトを個別に設定することも可能です。

import requests

url = 'https://httpbin.org/delay/5' # 5秒遅延するエンドポイント

try:
    # タイムアウトを3秒に設定 (接続・読み取り共通)
    response = requests.get(url, timeout=3)
except requests.exceptions.Timeout:
    print("リクエストがタイムアウトしました (共通設定)")

try:
    # 接続タイムアウトを1秒、読み取りタイムアウトを4秒に設定
    # 接続タイムアウト: サーバーへの接続確立までの時間
    # 読み取りタイムアウト: 接続後、サーバーからの最初のバイト受信までの時間 (チャンク間の時間ではない)
    response = requests.get(url, timeout=(1, 4))
except requests.exceptions.ConnectTimeout:
    print("接続がタイムアウトしました")
except requests.exceptions.ReadTimeout:
    print("データの読み取りがタイムアウトしました")

try:
    # タイムアウトなし (非推奨: 無限に待機する可能性)
    response_no_timeout = requests.get(url, timeout=None)
    print("タイムアウトなしで完了 (遅延後)")
except requests.exceptions.RequestException as e:
    print(f"タイムアウトなしでもエラー発生の可能性: {e}")

外部へのリクエストには常に適切なタイムアウトを設定することが強く推奨されます。デフォルトではタイムアウトは設定されていません。

リダイレクトの制御

デフォルトではRequestsはリダイレクト (3xxステータスコード) に自動的に追従します。allow_redirects引数でこの挙動を制御できます。

import requests

url = 'https://httpbin.org/redirect/3' # 3回リダイレクトするエンドポイント

# デフォルト (リダイレクト許可)
response_allow = requests.get(url)
print(f"リダイレクト許可時の最終URL: {response_allow.url}")
print(f"リダイレクト履歴: {[r.url for r in response_allow.history]}")
print(f"リダイレクト許可時のステータス: {response_allow.status_code}") # 通常はリダイレクト先の200

# リダイレクト禁止
response_deny = requests.get(url, allow_redirects=False)
print(f"\nリダイレクト禁止時のURL: {response_deny.url}") # 元のURL
print(f"リダイレクト履歴: {response_deny.history}") # 空のリスト
print(f"リダイレクト禁止時のステータス: {response_deny.status_code}") # 302 Found など
print(f"Locationヘッダー: {response_deny.headers.get('Location')}") # リダイレクト先URL

SSL証明書の検証制御 🔒

デフォルトではRequestsはHTTPS接続時にサーバーのSSL証明書を検証します。verify引数でこの挙動を変更できます。

import requests

url_secure = 'https://example.com' # 通常は検証OK
url_self_signed = 'https://self-signed.badssl.com/' # 自己署名証明書 (検証失敗するはず)

# デフォルト (検証有効)
try:
    response_verify_ok = requests.get(url_secure) # 通常成功
    print(f"検証OK: {url_secure} - ステータス: {response_verify_ok.status_code}")

    response_verify_fail = requests.get(url_self_signed) # ここで例外が発生するはず
except requests.exceptions.SSLError as e:
    print(f"SSL検証エラー (期待通り): {e}")

# 検証を無効にする (非推奨: セキュリティリスクあり)
import warnings
from urllib3.exceptions import InsecureRequestWarning

# InsecureRequestWarning を抑制 (自己責任で)
warnings.simplefilter('ignore', InsecureRequestWarning)

try:
    response_no_verify = requests.get(url_self_signed, verify=False)
    print(f"\n検証無効時のステータス ({url_self_signed}): {response_no_verify.status_code} (警告抑制)")
except requests.exceptions.RequestException as e:
    print(f"検証無効でもエラー発生の可能性: {e}")

# 特定のCAバンドルファイルまたはディレクトリを指定
# try:
#     # 例: カスタムCA証明書ファイルへのパス
#     response_custom_ca = requests.get(url_secure, verify='/path/to/custom/ca.pem')
#     print(f"\nカスタムCA使用時のステータス: {response_custom_ca.status_code}")
# except requests.exceptions.RequestException as e:
#     print(f"カスタムCAエラー: {e}")

verify=False は中間者攻撃のリスクを高めるため、絶対にプロダクション環境で使用しないでください。テスト環境や、信頼できる内部ネットワークでのみ限定的に使用すべきです。必要な場合は、代わりに正しいCA証明書を指定 (verify='/path/to/cert') するか、システムの証明書ストアを更新してください。

クライアント証明書を使用する場合は cert 引数を使います。

# cert引数にクライアント証明書ファイルと秘密鍵ファイルのパスをタプルで指定
# response = requests.get('https://example.com', cert=('/path/to/client.crt', '/path/to/client.key'))
# 証明書と鍵が1つのファイルにまとまっている場合は、そのファイルのパスを文字列で指定
# response = requests.get('https://example.com', cert='/path/to/client.pem')

プロキシ設定 🌐

proxies引数にプロキシサーバーのアドレスを辞書で指定します。

import requests

# プロキシサーバーの情報を設定
proxies = {
  'http': 'http://10.10.1.10:3128',    # HTTPリクエスト用のプロキシ
  'https': 'http://user:password@10.10.1.10:1080', # HTTPSリクエスト用のプロキシ (認証情報付き)
  # 'https': 'socks5://user:password@host:port' # SOCKSプロキシ (別途 `requests[socks]` のインストールが必要)
}

# 環境変数 HTTP_PROXY, HTTPS_PROXY が設定されている場合、Requestsは自動的にそれを使用します。
# proxies引数で明示的に指定すると、環境変数の設定は上書きされます。

try:
    # プロキシ経由でリクエスト
    response = requests.get('https://httpbin.org/ip', proxies=proxies)
    print("プロキシ経由で取得したIP情報:")
    print(response.json()) # プロキシサーバーのIPが表示されるはず

    # プロキシを使わない場合は空の辞書を渡すか、引数を省略
    # response_no_proxy = requests.get('https://httpbin.org/ip', proxies={})
except requests.exceptions.ProxyError as e:
    print(f"プロキシエラーが発生しました: {e}")
except requests.exceptions.RequestException as e:
    print(f"リクエストエラー（プロキシ関連含む）: {e}")

SOCKSプロキシを使用するには、追加のライブラリが必要です。pip install requests[socks] を実行してください。

リクエスト中に発生する可能性のある様々なエラーを適切に処理します。

基本的なHTTPエラーのチェック

response.raise_for_status() メソッドを使うのが最も簡単です。4xx (クライアントエラー) または 5xx (サーバーエラー) のステータスコードの場合に requests.exceptions.HTTPError を発生させます。

import requests

urls_to_check = [
    'https://httpbin.org/status/200', # OK
    'https://httpbin.org/status/404', # Not Found
    'https://httpbin.org/status/500'  # Internal Server Error
]

for url in urls_to_check:
    try:
        response = requests.get(url, timeout=5)
        # ステータスコードが4xxか5xxなら例外発生
        response.raise_for_status()
        print(f"{url}: 成功 (ステータス {response.status_code})")
    except requests.exceptions.HTTPError as http_err:
        print(f"{url}: HTTPエラーが発生しました: {http_err} (ステータス {http_err.response.status_code})")
    except requests.exceptions.ConnectionError as conn_err:
        print(f"{url}: 接続エラーが発生しました: {conn_err}")
    except requests.exceptions.Timeout as timeout_err:
        print(f"{url}: タイムアウトエラーが発生しました: {timeout_err}")
    except requests.exceptions.RequestException as req_err:
        # 上記以外のRequests関連のエラー (リダイレクトが多すぎるなど)
        print(f"{url}: リクエストエラーが発生しました: {req_err}")

具体的な例外クラス

requests.exceptions モジュールには、より具体的なエラー状況に対応する例外クラスが定義されています。これらを個別にキャッチすることで、エラーの種類に応じた処理を行うことができます。

エラーハンドリングを行う際は、より具体的な例外からキャッチし、最後に汎用的な requests.exceptions.RequestException をキャッチするのが一般的です。

より特殊なユースケースに対応するための機能です。

ストリーミングダウンロード (大きなファイル)

stream=True を指定すると、レスポンスボディをすぐにダウンロードせず、必要な時に読み込むことができます。これにより、メモリ使用量を抑えながら大きなファイルを扱うことが可能です。

import requests
import shutil # ファイル操作用

url = 'https://speed.hetzner.de/100MB.bin' # 例: 大きなファイル
file_path = 'downloaded_100MB.bin'

try:
    # stream=Trueでリクエスト
    with requests.get(url, stream=True, timeout=60) as r: # タイムアウトも適切に設定
        r.raise_for_status() # エラーチェック

        # レスポンスヘッダーからファイルサイズを取得 (存在すれば)
        total_size = int(r.headers.get('content-length', 0))
        print(f"ダウンロード開始: {file_path} ({total_size} バイト)")

        # ファイルにチャンクごとに書き込む
        chunk_size = 8192 # 8KBずつ読み込む
        downloaded_size = 0
        with open(file_path, 'wb') as f:
            # iter_contentを使ってチャンクを取得
            for chunk in r.iter_content(chunk_size=chunk_size):
                # chunkには最大chunk_sizeバイトのデータが含まれる (Noneの場合もある)
                if chunk:
                    f.write(chunk)
                    downloaded_size += len(chunk)
                    # 進捗表示 (例)
                    progress = (downloaded_size / total_size) * 100 if total_size > 0 else 0
                    print(f"\rダウンロード中: {downloaded_size}/{total_size} バイト ({progress:.2f}%)", end="")

        print("\nダウンロード完了！")

    # shutilを使ってファイルコピーを行うシンプルな例 (メモリに読み込まない)
    # with requests.get(url, stream=True) as r:
    #     r.raise_for_status()
    #     with open(file_path + ".shutil", 'wb') as f:
    #         shutil.copyfileobj(r.raw, f) # response.raw は低レベルなソケットアクセス
    # print("shutilでのコピー完了！")


except requests.exceptions.RequestException as e:
    print(f"\nダウンロード中にエラーが発生しました: {e}")
except Exception as e:
    print(f"\n予期せぬエラー: {e}")

stream=True を使用した場合、レスポンスのコンテンツにアクセスした後、必ず response.close() を呼び出すか、with 文を使用して接続を解放することが重要です。iter_content() や iter_lines() を使うと、コンテンツを効率的に処理できます。response.raw は、より低レベルなアクセスを提供しますが、通常は iter_content が推奨されます。

リクエストフック (Hooks)

リクエスト処理の特定のポイントでカスタム関数を呼び出すことができます。主にデバッグやリクエスト/レスポンスのロギング、ヘッダーの動的な変更などに使用されます。

import requests

def print_url(response, *args, **kwargs):
    """レスポンスオブジェクトを受け取り、URLを出力するフック関数"""
    print(f"フック: リクエスト完了 - URL: {response.url}, Status: {response.status_code}")

def add_custom_header(request, *args, **kwargs):
    """準備されたリクエストオブジェクトを受け取り、ヘッダーを追加するフック関数"""
    request.headers['X-Request-Hooked'] = 'True'
    print(f"フック: リクエスト準備完了 - {request.method} {request.url}")
    # 注意: このフックは PreparedRequest オブジェクトを操作する
    return request # フック関数は None または PreparedRequest を返す必要がある

# フックを辞書で定義
hooks = {
    'response': [print_url], # レスポンス受信後に呼び出されるフック (複数の関数をリストで指定可)
    'request': add_custom_header # リクエスト準備完了後、送信前に呼び出されるフック (単一の関数)
}

# hooks引数を指定してリクエスト
response = requests.get('https://httpbin.org/get', hooks=hooks)

print("\nレスポンスヘッダー (サーバー視点):")
# サーバー側で認識されたヘッダーを確認 (X-Request-Hooked があるはず)
# print(response.json()['headers'])

# セッションオブジェクト全体にフックを設定することも可能
# s = requests.Session()
# s.hooks['response'].append(print_url)
# s.get('https://httpbin.org/delay/1')

利用可能なフックポイントは 'response' (レスポンス受信後) と 'request' (リクエスト準備完了後) です。

フック関数は慎重に設計する必要があります。特にレスポンスフックは、ストリーミングレスポンスの場合、コンテンツが消費される前に呼び出されます。フック内でのエラーはリクエスト全体の失敗につながる可能性があります。