はじめに
requestsはPythonで最も人気のあるHTTPライブラリです。APIとの通信やWebスクレイピングに欠かせません。
インストール
pip install requests
基本的なリクエスト
GETリクエスト
import requests
# 基本的なGET
response = requests.get("https://api.github.com/users/octocat")
print(response.status_code) # 200
print(response.text) # レスポンスボディ（文字列）
print(response.json()) # JSONをパース
print(response.headers) # レスポンスヘッダー
クエリパラメータ
# URLに直接書く
response = requests.get("https://api.example.com/search?q=python&page=1")
# params引数を使う（推奨）
params = {
"q": "python",
"page": 1,
"sort": "stars"
}
response = requests.get("https://api.github.com/search/repositories", params=params)
print(response.url) # 自動的にURLエンコードされる
POSTリクエスト
# フォームデータ
data = {
"username": "user1",
"password": "pass123"
}
response = requests.post("https://api.example.com/login", data=data)
# JSONデータ
json_data = {
"title": "New Post",
"body": "Content here",
"userId": 1
}
response = requests.post("https://jsonplaceholder.typicode.com/posts", json=json_data)
print(response.json())
その他のHTTPメソッド
# PUT（更新）
response = requests.put("https://api.example.com/users/1", json={"name": "新しい名前"})
# PATCH（部分更新）
response = requests.patch("https://api.example.com/users/1", json={"email": "new@example.com"})
# DELETE（削除）
response = requests.delete("https://api.example.com/users/1")
# HEAD（ヘッダーのみ取得）
response = requests.head("https://example.com")
ヘッダーの設定
headers = {
"User-Agent": "MyApp/1.0",
"Accept": "application/json",
"Authorization": "Bearer token123"
}
response = requests.get("https://api.example.com/data", headers=headers)
認証
Basic認証
from requests.auth import HTTPBasicAuth
response = requests.get(
"https://api.example.com/private",
auth=HTTPBasicAuth("username", "password")
)
# 省略形
response = requests.get(
"https://api.example.com/private",
auth=("username", "password")
)
Bearer Token
headers = {"Authorization": "Bearer YOUR_TOKEN"}
response = requests.get("https://api.example.com/me", headers=headers)
タイムアウトの設定
try:
# 接続タイムアウト3秒、読み込みタイムアウト10秒
response = requests.get("https://api.example.com/data", timeout=(3, 10))
except requests.exceptions.Timeout:
print("リクエストがタイムアウトしました")
エラーハンドリング
import requests
from requests.exceptions import HTTPError, ConnectionError, Timeout, RequestException
try:
response = requests.get("https://api.example.com/data", timeout=5)
response.raise_for_status() # 4xx/5xxでHTTPErrorを発生
data = response.json()
except HTTPError as e:
print(f"HTTPエラー: {e.response.status_code}")
except ConnectionError:
print("接続エラー: サーバーに接続できません")
except Timeout:
print("タイムアウト: 応答がありません")
except RequestException as e:
print(f"リクエストエラー: {e}")
セッションの使用
複数のリクエストで設定を共有できます。
session = requests.Session()
# 共通のヘッダーを設定
session.headers.update({
"User-Agent": "MyApp/1.0",
"Authorization": "Bearer token123"
})
# Cookieも自動的に保持される
response = session.get("https://api.example.com/login")
response = session.get("https://api.example.com/profile") # ログイン状態が維持される
# セッションを閉じる
session.close()
# withを使う方法
with requests.Session() as session:
session.headers.update({"Authorization": "Bearer token123"})
response = session.get("https://api.example.com/data")
ファイルのアップロード
# 単一ファイル
with open("image.png", "rb") as f:
files = {"file": f}
response = requests.post("https://api.example.com/upload", files=files)
# ファイル名とContent-Typeを指定
files = {
"file": ("report.pdf", open("document.pdf", "rb"), "application/pdf")
}
response = requests.post("https://api.example.com/upload", files=files)
# 複数ファイル
files = [
("images", ("image1.png", open("1.png", "rb"), "image/png")),
("images", ("image2.png", open("2.png", "rb"), "image/png")),
]
response = requests.post("https://api.example.com/upload", files=files)
ファイルのダウンロード
# 小さいファイル
response = requests.get("https://example.com/file.pdf")
with open("downloaded.pdf", "wb") as f:
f.write(response.content)
# 大きいファイル（ストリーミング）
response = requests.get("https://example.com/large_file.zip", stream=True)
with open("large_file.zip", "wb") as f:
for chunk in response.iter_content(chunk_size=8192):
f.write(chunk)
リトライの実装
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
session = requests.Session()
# リトライ設定
retry_strategy = Retry(
total=3, # 最大リトライ回数
backoff_factor=1, # リトライ間隔
status_forcelist=[429, 500, 502, 503, 504], # リトライするステータスコード
)
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount("http://", adapter)
session.mount("https://", adapter)
response = session.get("https://api.example.com/data")
実践例：GitHub API クライアント
import requests
from dataclasses import dataclass
from typing import Optional, List
@dataclass
class GitHubUser:
login: str
name: Optional[str]
public_repos: int
followers: int
class GitHubClient:
BASE_URL = "https://api.github.com"
def __init__(self, token: Optional[str] = None):
self.session = requests.Session()
self.session.headers.update({
"Accept": "application/vnd.github.v3+json",
"User-Agent": "Python-GitHub-Client"
})
if token:
self.session.headers["Authorization"] = f"token {token}"
def get_user(self, username: str) -> GitHubUser:
"""ユーザー情報を取得"""
response = self.session.get(f"{self.BASE_URL}/users/{username}")
response.raise_for_status()
data = response.json()
return GitHubUser(
login=data["login"],
name=data.get("name"),
public_repos=data["public_repos"],
followers=data["followers"]
)
def search_repos(self, query: str, sort: str = "stars", limit: int = 10) -> List[dict]:
"""リポジトリを検索"""
params = {
"q": query,
"sort": sort,
"per_page": limit
}
response = self.session.get(f"{self.BASE_URL}/search/repositories", params=params)
response.raise_for_status()
return response.json()["items"]
def close(self):
self.session.close()
# 使用例
client = GitHubClient()
try:
user = client.get_user("octocat")
print(f"{user.login}: {user.public_repos} repos, {user.followers} followers")
repos = client.search_repos("python web framework", limit=5)
for repo in repos:
print(f"- {repo['full_name']}: ⭐ {repo['stargazers_count']}")
finally:
client.close()
まとめ
requests.get/post/put/deleteで各HTTPメソッドを実行paramsでクエリパラメータ、jsonでJSONボディを送信response.json()でJSONをパースtimeoutで接続タイムアウトを設定Sessionで設定を共有し、Cookieを維持raise_for_status()でHTTPエラーを例外として処理
次回はpandasでデータ分析を学びます。