19.

Python で HTTP リクエスト|requests・urllib・httpx と curl の対応

編集
この記事の要点
  • Python から HTTP を叩くなら requestspip install requests で入る
  • 標準ライブラリだけで済ませたいときは urllib.request。非同期や HTTP/2 が要るなら httpx
  • JSON を送るのは json=、フォーム送信は data=両者で Content-Type が変わる
  • timeout を必ず指定する。省略すると応答が無いとき永久に待つ
  • ステータスは自動で例外にならない。raise_for_status() を呼ぶ

curl コマンドとの対応

curlrequests
curl URLrequests.get(url)
curl -X POST -d "a=1"requests.post(url, data={"a": 1})
curl -H "Content-Type: application/json" でボディを送るrequests.post(url, json={...})
curl -H "Authorization: Bearer X"headers={"Authorization": "Bearer X"}
curl -u user:passauth=("user", "pass")
curl -L既定でリダイレクト追従(切るなら allow_redirects=False
curl --max-time 10timeout=10
curl -F "file=@x.png"files={"file": open("x.png", "rb")}
curl -kverify=False(本番では使わない)

GET

import requests

res = requests.get(
    "https://api.example.com/users",
    params={"page": 2, "limit": 50},        # ?page=2&limit=50 に組み立てられる
    headers={"Authorization": "Bearer TOKEN"},
    timeout=10,                             # 必ず指定する
)

res.raise_for_status()                      # 4xx / 5xx なら例外
data = res.json()                           # JSON を dict に
print(res.status_code)                      # 200
print(res.headers["Content-Type"])
print(len(data))

params に辞書を渡せば URL エンコードは自動です。日本語や記号を自分で組み立てる必要はありません。

POST(JSON とフォーム)

# JSON で送る(Content-Type: application/json が自動で付く)
res = requests.post(
    "https://api.example.com/users",
    json={"name": "田中", "age": 30},
    timeout=10,
)

# フォーム送信(application/x-www-form-urlencoded)
res = requests.post(url, data={"name": "田中"}, timeout=10)

# 生のボディを送る
res = requests.post(url, data="raw text".encode("utf-8"),
                    headers={"Content-Type": "text/plain"}, timeout=10)

# ファイルアップロード(multipart/form-data)
with open("photo.png", "rb") as f:
    res = requests.post(url, files={"file": ("photo.png", f, "image/png")},
                        data={"title": "写真"}, timeout=10)

json=data= を取り違えるのが最も多い失敗です。API 側が JSON を期待しているのに data= で辞書を渡すと、フォーム形式で送られて 400 が返ります。

レスポンスの中身

res = requests.get(url, timeout=10)

print(res.status_code)   # 200
print(res.ok)            # 400 未満なら True
print(res.text)          # 文字列(res.encoding に従ってデコード)
print(res.content)       # bytes(画像・PDF などはこちら)
print(res.json())        # JSON をパース。失敗すると JSONDecodeError
print(res.url)           # 最終的にアクセスした URL
print(res.elapsed)       # かかった時間

# 文字化けするときはエンコーディングを明示
res.encoding = "utf-8"
print(res.text)

詳しくは Responseオブジェクトの中身の確認 を参照してください。

エラー処理とタイムアウト

import requests

try:
    res = requests.get(url, timeout=(3.0, 10.0))   # (接続, 読み取り) を別々に
    res.raise_for_status()
    data = res.json()
except requests.Timeout:
    print("時間内に応答がありませんでした")
except requests.ConnectionError:
    print("接続できませんでした(DNS / ネットワーク)")
except requests.HTTPError as e:
    print(f"HTTP エラー {e.response.status_code}: {e.response.text[:200]}")
except requests.JSONDecodeError:
    print("JSON として解釈できませんでした")
except requests.RequestException as e:
    print(f"その他の失敗: {e}")

timeout を省略すると既定では待ち続けます。相手が応答しないとプログラムが止まったままになるため、必ず指定します。

Session で接続を使い回す

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

session = requests.Session()
session.headers.update({"Authorization": "Bearer TOKEN"})   # 毎回付く

retry = Retry(total=3, backoff_factor=1,
              status_forcelist=[429, 500, 502, 503, 504])
session.mount("https://", HTTPAdapter(max_retries=retry))

for page in range(1, 6):
    res = session.get(url, params={"page": page}, timeout=10)
    res.raise_for_status()

session.close()

同じホストに何度もアクセスするなら Session を使います。TCP 接続とヘッダー・Cookie が引き継がれるので速く、再試行の設定も 1 か所にまとめられます。Cookie の扱いは cookieの値の設定と取得 を参照してください。

標準ライブラリだけで済ませる

import json
import urllib.parse
import urllib.request

# GET
url = "https://api.example.com/users?" + urllib.parse.urlencode({"page": 2})
req = urllib.request.Request(url, headers={"User-Agent": "my-app/1.0"})
with urllib.request.urlopen(req, timeout=10) as res:
    data = json.loads(res.read().decode("utf-8"))

# POST(JSON)
body = json.dumps({"name": "田中"}).encode("utf-8")
req = urllib.request.Request(url, data=body, method="POST",
                             headers={"Content-Type": "application/json"})
with urllib.request.urlopen(req, timeout=10) as res:
    print(res.status)

追加インストールができない環境ではこちらを使います。urlopen は 4xx / 5xx で urllib.error.HTTPError を投げる点が requests と違います。

httpx(非同期・HTTP/2)

import asyncio
import httpx

async def main(urls):
    async with httpx.AsyncClient(timeout=10) as client:
        results = await asyncio.gather(*(client.get(u) for u in urls))
    for r in results:
        print(r.status_code, r.url)

asyncio.run(main(["https://example.com", "https://example.org"]))

API はほぼ requests と同じで、複数のリクエストを同時に投げたい場合に効果があります。

安全面の注意

  • verify=False は証明書検証を無効にする。本番では使わない(社内証明書なら verify に CA ファイルのパスを渡す)
  • トークンをソースに直書きせず、環境変数(os.environ["API_TOKEN"])から読む
  • ユーザー入力をそのまま URL にしない。アクセス先を限定する
  • 取得したコンテンツのサイズ上限を決める(stream=True で分割読み込み)

関連

編集
Post Share
子ページ

子ページはありません

同階層のページ
  1. 基本的なルール
  2. 変数
  3. 演算子
  4. 標準ライブラリ
  5. 外部ライブラリ
  6. 制御構文
  7. リスト(配列)
  8. タプル
  9. セット
  10. 辞書(dict)
  11. クラスとメソッド
  12. 継承の概念と必要性
  13. 継承の構文
  14. コンストラクタ
  15. cookieの値の設定と取得
  16. 例外処理
  17. 例外を文字列で出力する方法
  18. httpリクエスト(curl)をする方法
  19. Responseオブジェクトの中身の確認
  20. 変数が空かどうか判定する方法
  21. タイムゾーンの設定と現在日時の取得と文字列化
  22. シングルクォーテーションとダブルクォーテーションの違い

最近更新/作成されたページ