外部APIと連携する実装で、認証やタイムアウト、エラー時の保存・再試行などでつまずくことは多いです。本記事では「業務で再現可能かつ安全に」APIを叩くための基本ワークフローを、短いコード例とチェックリストで整理します。前提となるPythonの基本（関数・ファイルI/O・例外処理）は知っていることを想定しています。
問題設定とユースケース
想定ユースケース：SaaSやモデルAPIからページネーションでデータを取得し、CSVに追記して業務で利用する。要点は認証情報を安全に管理し、接続再利用・タイムアウト・応答検証・失敗時の保存・再実行用メタデータを用意することです。
前提環境と秘密情報の管理
環境変数と .env の使い方
シークレットはコードに直書きせず、環境変数で管理します。ローカルでは .env を使い、CI／本番ではシークレットストアを利用するのが実務的です。以下は読み取りのテンプレートです。
# requirements: python-dotenv（ローカル用）
from dotenv import load_dotenv
import os
load_dotenv() # ローカルでのみ.envを読み込む
API_KEY = os.environ.get('MY_API_KEY')
if not API_KEY:
raise RuntimeError('MY_API_KEY が設定されていません')
ポイント：
- .env は .gitignore に入れる
- CI・クラウド環境では環境変数かシークレットマネージャを使用
- ログにシークレットを出力しない
同期リクエスト設計：Session とタイムアウト
requests.Session を使い接続を再利用します。タイムアウトを必ず設定し、短め（例：connect 3s / read 10s）を推奨します。
import requests
def create_session(api_key: str) -> requests.Session:
s = requests.Session()
s.headers.update({
'Authorization': f'Bearer {api_key}',
'Accept': 'application/json',
'User-Agent': 'manageai/1.0'
})
return s
# 利用例
# session = create_session(API_KEY)
# resp = session.get(url, timeout=(3, 10))
エラー分類とハンドリング方針
主に次の3分類で考えます。
| 分類 | 例 | 対応方針 |
|---|---|---|
| HTTPエラー | 4xx, 5xx | 4xxは再試行前に内容確認。401/403は認証、403は権限確認。5xxは短期の再試行か別ルート検討（第124回参照）。 |
| ネットワーク | タイムアウト・接続遮断 | 短時間の再試行＋バックオフ。ただし即時リトライは避ける（第124回参照）。ログ保存して後段で再処理可能に。 |
| パース/スキーマ | JSONDecodeError・期待フィールドがない | 受信内容を保存して手動確認。スキーマチェックで早期に弾く。 |
短いハンドリングテンプレート
import json
from requests.exceptions import RequestException
def safe_get(session, url, params=None, timeout=(3,10)):
try:
r = session.get(url, params=params, timeout=timeout)
r.raise_for_status()
return r.json()
except RequestException as e:
# ネットワークやHTTPエラーをログに残す
raise
except json.JSONDecodeError:
# レスポンスの保存とアラート
raise
レート制限とポリシー
APIのレート制限は運用側で尊重します。検知方法は主にレスポンスヘッダ（Retry-After など）や429ステータスです。即時リトライ禁止、指数バックオフ＋最大待機時間を設定します。
- 429 や Retry-After があればその値に従う
- ヘッダがなければ指数バックオフ（例：1s, 2s, 4s）を上限で止める
- 長時間の処理はジョブ化して非同期に回す（次回：非同期/httpx予定）
ページネーションとストリーミング応答の扱い
ページネーションはループで確実に次ページを取得し、途中で失敗した場合は現在の取得位置をメタデータとして保存して再開できるようにします。
import csv
from typing import Dict, Any
def fetch_all_paginated(session, base_url, params=None, save_row_fn=None):
page = 1
while True:
params = params or {}
params.update({'page': page})
data = safe_get(session, base_url, params=params)
items = data.get('items', [])
for it in items:
if save_row_fn:
save_row_fn(it)
if not data.get('has_more'):
break
page += 1
CSVへ追記する save_row_fn の一例：
def save_to_csv(file_path: str, row: Dict[str, Any]):
header = ['id', 'name', 'value']
write_header = not os.path.exists(file_path)
with open(file_path, 'a', newline='', encoding='utf-8') as f:
writer = csv.DictWriter(f, fieldnames=header)
if write_header:
writer.writeheader()
writer.writerow({k: row.get(k) for k in header})
応答検証と簡易スキーマチェック
外部依存のため応答が変わることがあります。起きがちな問題はキー名の欠落や型の変更です。軽いバリデーションを入れて早期に検出します。
def validate_item_schema(item: dict) -> bool:
# 必要最小限のチェック
required = ['id', 'name']
for k in required:
if k not in item:
return False
return True
応答の保存と冪等性
失敗時の再実行に備え、取得済みのメタデータを残します。idempotency key を使うと二重登録を防げます。
- 取得済みログ（id・timestamp・page）を小さなファイルやDBに残す
- 登録APIには idempotency キーを付与する（ヘッダやbody）
- 失敗レスポンスは原文（JSON）をファイル保存して手動調査できるようにする
テストとCIでのモック
外部APIに対するテストはモックを使います。requests-mock や responses ライブラリを使い、HTTPステータス・時間切れ・不正JSONなどを再現します。CIでは本物のAPIキーは使わないでください。
運用時の観測ポイントとログ設計
| 観測ポイント | 具体例 |
|---|---|
| 成功率 | API呼び出し当たりの200応答率 |
| 平均遅延 | connect/read の時間 |
| レート制限検出 | 429発生回数、Retry-After付き回数 |
| 異常レスポンス保存件数 | JSON解析失敗やスキーマ違反の件数 |
サンプル入出力（例）
| APIレスポンス（itemsの例） | CSV出力行 |
|---|---|
| {“id”: 123, “name”: “Widget A”, “value”: 9.5} | 123,Widget A,9.5 |
チェックリスト（実装・運用の必須項目）
- APIキーは環境変数で管理している（.env は .gitignore）
- Session を使って接続再利用している
- connect/read timeout を必ず設定している
- HTTPステータスごとのハンドリング方針がある
- レート制限検知（429 / Retry-After）に従う実装がある
- 取得途中で失敗した際に再開できるメタデータを保存している
- 外部APIをモックしてCIでテストしている
まとめ
本稿では認証管理、Sessionによる接続再利用、タイムアウト、エラー分類、ページネーション処理、簡易スキーマ検証、応答保存と冪等性、テスト方針までを実務指向でまとめました。耐障害性の再試行・バックオフの詳細は第124回を参照してください（内部検索：第124回 再試行・バックオフ）。また、ページネーションサンプルは第113回/第122回の記事とも関連があります（検索：第113回、第122回）。
次の一歩（5分で試せる手順）
- ローカルに .env を作り、MY_API_KEY=あなたのキー を記入する
- このリポジトリに requests と python-dotenv をインストールする（pip install requests python-dotenv）
- 上記の create_session と fetch_all_paginated、save_to_csv をコピーして実行してみる（テスト用のモックAPIでも可）
- CSVが追記されることを確認して、エラー時にJSONファイルが残るように例外処理を追加する
次回は非同期／ストリーミングを想定した httpx / asyncio を扱い、長時間接続や大量データ受信の実務上の注意を紹介します（予告）。