タイトル: httpリクエスト(curl)をする方法
SEOタイトル: Python で HTTP リクエスト|requests・urllib・httpx と curl の対応
| この記事の要点 |
|
curl コマンドとの対応
| curl | requests |
|---|---|
curl URL | requests.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:pass | auth=("user", "pass") |
curl -L | 既定でリダイレクト追従(切るなら allow_redirects=False) |
curl --max-time 10 | timeout=10 |
curl -F "file=@x.png" | files={"file": open("x.png", "rb")} |
curl -k | verify=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で分割読み込み)
関連
- 文法 — 親カテゴリ
- Responseオブジェクトの中身の確認
- cookieの値の設定と取得
- 外部ライブラリ —
requestsの導入 - 例外処理
- Webスクレイピング