Python自動化マスター第7編：API活用とデータ収集
Python Automation Master Part 7: API Integration and Data Collection
はじめに：APIが開くデータの世界
現代のWebサービスの核心はAPI（Application Programming Interface）です。APIを通じて、天気情報、株価、ニュース、ソーシャルメディアデータなど、様々な情報をプログラム的に収集・活用できます。Pythonのrequestsライブラリを使用すると、このようなAPI呼び出しを非常に簡単に処理できます。
今回の編では、REST APIの基本概念から始めて、様々な認証方式、実践的なAPI活用法、そして効率的なデータ収集戦略まで体系的に学習します。
1. API基礎：REST APIの理解
1.1 APIとは何か？
APIは異なるソフトウェア間の通信を可能にするインターフェースです。レストランに例えると、APIはお客様（クライアント）と厨房（サーバー）の間のウェイター役を果たします。お客様がメニューを注文するとウェイターが厨房に伝え、料理が完成するとお客様に届けるようなものです。
# APIリクエストの基本構造
import requests
# 1. リクエストを送信
response = requests.get("https://api.example.com/data")
# 2. レスポンスを受け取る
if response.status_code == 200:
data = response.json()
print(data)
else:
print(f"エラー発生: {response.status_code}")
1.2 REST APIの核心概念
REST（Representational State Transfer）はWeb API設計の標準アーキテクチャです。
# HTTPメソッド別の用途
"""
GET : データ取得 (Read)
POST : データ作成 (Create)
PUT : データ全体更新 (Update)
PATCH : データ部分更新 (Partial Update)
DELETE : データ削除 (Delete)
"""
import requests
base_url = "https://api.example.com"
# GET - データ取得
response = requests.get(f"{base_url}/users")
# POST - データ作成
new_user = {"name": "田中太郎", "email": "tanaka@example.com"}
response = requests.post(f"{base_url}/users", json=new_user)
# PUT - データ更新
updated_user = {"name": "田中太郎", "email": "newemail@example.com"}
response = requests.put(f"{base_url}/users/1", json=updated_user)
# DELETE - データ削除
response = requests.delete(f"{base_url}/users/1")
1.3 HTTPステータスコードの理解
# 主要HTTPステータスコード
status_codes = {
# 2xx: 成功
200: "OK - リクエスト成功",
201: "Created - 作成成功",
204: "No Content - 成功（レスポンス本文なし）",
# 3xx: リダイレクション
301: "Moved Permanently - 永久移動",
302: "Found - 一時移動",
# 4xx: クライアントエラー
400: "Bad Request - 不正なリクエスト",
401: "Unauthorized - 認証が必要",
403: "Forbidden - アクセス拒否",
404: "Not Found - リソースなし",
429: "Too Many Requests - リクエスト回数超過",
# 5xx: サーバーエラー
500: "Internal Server Error - サーバー内部エラー",
502: "Bad Gateway - ゲートウェイエラー",
503: "Service Unavailable - サービス利用不可"
}
def handle_response(response):
"""レスポンスステータスコードに応じた処理"""
code = response.status_code
if 200 <= code < 300:
print(f"成功: {status_codes.get(code, '不明な成功コード')}")
return response.json() if response.content else None
elif 400 <= code < 500:
print(f"クライアントエラー: {status_codes.get(code, '不明なエラー')}")
return None
elif 500 <= code < 600:
print(f"サーバーエラー: {status_codes.get(code, '不明なサーバーエラー')}")
return None
2. requestsライブラリ完全ガイド
2.1 基本的な使い方
import requests
# インストール: pip install requests
# GET リクエストの基本
response = requests.get("https://api.github.com/users/octocat")
print(response.status_code) # 200
print(response.headers) # レスポンスヘッダー
print(response.text) # テキスト形式のレスポンス
print(response.json()) # JSONパースされた辞書
# URLパラメータの渡し方
params = {
"q": "python",
"sort": "stars",
"order": "desc"
}
response = requests.get(
"https://api.github.com/search/repositories",
params=params
)
# 実際のリクエストURL: https://api.github.com/search/repositories?q=python&sort=stars&order=desc
2.2 ヘッダーとタイムアウト設定
import requests
# カスタムヘッダー設定
headers = {
"User-Agent": "MyApp/1.0",
"Accept": "application/json",
"Content-Type": "application/json"
}
# タイムアウト設定（接続タイムアウト、読み取りタイムアウト）
try:
response = requests.get(
"https://api.example.com/data",
headers=headers,
timeout=(5, 30) # 接続: 5秒、読み取り: 30秒
)
except requests.exceptions.Timeout:
print("リクエストがタイムアウトしました。")
except requests.exceptions.ConnectionError:
print("接続に失敗しました。")
except requests.exceptions.RequestException as e:
print(f"リクエスト中にエラーが発生: {e}")
2.3 セッションとCookie管理
import requests
# セッションを使用（Cookie自動管理）
session = requests.Session()
# ログイン
login_data = {"username": "user", "password": "pass"}
session.post("https://example.com/login", data=login_data)
# ログイン後のリクエスト（Cookie自動送信）
response = session.get("https://example.com/dashboard")
# セッションヘッダーの設定（すべてのリクエストに適用）
session.headers.update({
"Authorization": "Bearer token123",
"User-Agent": "MyApp/1.0"
})
3. API認証方式
3.1 APIキー認証
import requests
# ヘッダーにAPIキーを含める
headers = {
"X-API-Key": "your_api_key_here"
}
response = requests.get("https://api.example.com/data", headers=headers)
# URLパラメータにAPIキーを含める
params = {
"api_key": "your_api_key_here",
"query": "search_term"
}
response = requests.get("https://api.example.com/search", params=params)
3.2 Bearer Token認証
import requests
# Bearer Token認証
access_token = "your_access_token"
headers = {
"Authorization": f"Bearer {access_token}"
}
response = requests.get(
"https://api.example.com/protected/resource",
headers=headers
)
3.3 OAuth 2.0認証
import requests
class OAuth2Client:
"""OAuth 2.0クライアント"""
def __init__(self, client_id, client_secret, token_url):
self.client_id = client_id
self.client_secret = client_secret
self.token_url = token_url
self.access_token = None
def get_access_token(self):
"""アクセストークンを取得"""
data = {
"grant_type": "client_credentials",
"client_id": self.client_id,
"client_secret": self.client_secret
}
response = requests.post(self.token_url, data=data)
if response.status_code == 200:
token_data = response.json()
self.access_token = token_data["access_token"]
return self.access_token
return None
def make_request(self, url, method="GET", **kwargs):
"""認証済みリクエストを送信"""
if not self.access_token:
self.get_access_token()
headers = kwargs.get("headers", {})
headers["Authorization"] = f"Bearer {self.access_token}"
kwargs["headers"] = headers
response = requests.request(method, url, **kwargs)
# トークン期限切れの場合、再取得して再試行
if response.status_code == 401:
self.get_access_token()
headers["Authorization"] = f"Bearer {self.access_token}"
response = requests.request(method, url, **kwargs)
return response
4. 天気API活用例
import requests
from datetime import datetime
class WeatherAPI:
"""OpenWeatherMap API クラス"""
def __init__(self, api_key):
self.api_key = api_key
self.base_url = "https://api.openweathermap.org/data/2.5"
def get_current_weather(self, city):
"""現在の天気を取得"""
params = {
"q": city,
"appid": self.api_key,
"units": "metric", # 摂氏
"lang": "ja"
}
response = requests.get(f"{self.base_url}/weather", params=params)
if response.status_code == 200:
data = response.json()
return {
"city": data["name"],
"temperature": data["main"]["temp"],
"feels_like": data["main"]["feels_like"],
"humidity": data["main"]["humidity"],
"description": data["weather"][0]["description"],
"wind_speed": data["wind"]["speed"]
}
return None
def get_forecast(self, city, days=5):
"""天気予報を取得"""
params = {
"q": city,
"appid": self.api_key,
"units": "metric",
"lang": "ja",
"cnt": days * 8 # 3時間ごとのデータ
}
response = requests.get(f"{self.base_url}/forecast", params=params)
if response.status_code == 200:
data = response.json()
forecasts = []
for item in data["list"]:
forecasts.append({
"datetime": item["dt_txt"],
"temp": item["main"]["temp"],
"description": item["weather"][0]["description"]
})
return forecasts
return None
# 使用例
weather = WeatherAPI("YOUR_API_KEY")
current = weather.get_current_weather("Tokyo")
print(f"東京の現在の天気: {current['temperature']}°C, {current['description']}")
5. OpenAI API連携
import requests
class OpenAIClient:
"""OpenAI API クライアント"""
def __init__(self, api_key):
self.api_key = api_key
self.base_url = "https://api.openai.com/v1"
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
def chat_completion(self, messages, model="gpt-4", temperature=0.7):
"""チャット完了APIを呼び出す"""
url = f"{self.base_url}/chat/completions"
payload = {
"model": model,
"messages": messages,
"temperature": temperature
}
response = requests.post(url, headers=self.headers, json=payload)
if response.status_code == 200:
data = response.json()
return data["choices"][0]["message"]["content"]
else:
print(f"エラー: {response.status_code}")
print(response.text)
return None
def generate_summary(self, text, max_length=100):
"""テキストを要約"""
messages = [
{"role": "system", "content": "あなたは優秀な要約者です。与えられたテキストを簡潔に要約してください。"},
{"role": "user", "content": f"以下のテキストを{max_length}文字以内で要約してください:\n\n{text}"}
]
return self.chat_completion(messages)
def translate(self, text, target_language="日本語"):
"""テキストを翻訳"""
messages = [
{"role": "system", "content": f"あなたは優秀な翻訳者です。与えられたテキストを{target_language}に翻訳してください。"},
{"role": "user", "content": text}
]
return self.chat_completion(messages)
# 使用例
openai = OpenAIClient("YOUR_API_KEY")
# チャット
response = openai.chat_completion([
{"role": "user", "content": "Pythonの特徴を3つ教えてください。"}
])
print(response)
6. 効率的なデータ収集
import requests
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
class DataCollector:
"""効率的なデータ収集クラス"""
def __init__(self, max_workers=5, rate_limit=1.0):
self.max_workers = max_workers
self.rate_limit = rate_limit # 秒あたりのリクエスト数
self.last_request_time = 0
def _rate_limited_request(self, url, **kwargs):
"""レート制限付きリクエスト"""
elapsed = time.time() - self.last_request_time
wait_time = (1.0 / self.rate_limit) - elapsed
if wait_time > 0:
time.sleep(wait_time)
self.last_request_time = time.time()
return requests.get(url, **kwargs)
def collect_sequential(self, urls):
"""順次データ収集"""
results = []
for url in urls:
try:
response = self._rate_limited_request(url)
if response.status_code == 200:
results.append(response.json())
except Exception as e:
print(f"エラー ({url}): {e}")
return results
def collect_parallel(self, urls):
"""並列データ収集"""
results = []
with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
future_to_url = {
executor.submit(requests.get, url): url
for url in urls
}
for future in as_completed(future_to_url):
url = future_to_url[future]
try:
response = future.result()
if response.status_code == 200:
results.append(response.json())
except Exception as e:
print(f"エラー ({url}): {e}")
return results
# 使用例
collector = DataCollector(max_workers=5, rate_limit=2.0)
urls = [f"https://api.example.com/item/{i}" for i in range(100)]
data = collector.collect_parallel(urls)
まとめ
今回の編では、PythonでのAPI活用とデータ収集について学びました。
- REST API：HTTP メソッド（GET, POST, PUT, DELETE）の活用
- requestsライブラリ：Pythonで最も人気のあるHTTPクライアント
- 認証方式：APIキー、Bearer Token、OAuth 2.0
- 実践API：天気API、OpenAI APIなど
- 効率的な収集：レート制限、並列処理
APIは現代のソフトウェア開発において欠かせない技術です。様々なAPIを活用して自動化スクリプトを強化してください。次回の最終編では、スケジューリングと実践プロジェクトについて学びます。