外部APIの一時的な失敗に備える｜Pythonのリトライと指数バックオフ（requests）

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15
• 16
• 17
• 18
• 19
• 20
• 21
• 22
• 23
• 24
• 25
• 26
• 27
• 28
• 29
• 30
• 31
• 32
• 33
• 34
• 35
• 36
• 37
• 38

"""練習用のダミーAPIサーバー（ローカル専用）。
/flaky : 最初の FAIL_TIMES 回は 503、その後は 200 で JSON を返す
/notfound : いつも 404
"""
import json
import os
from http.server import BaseHTTPRequestHandler, HTTPServer

FAIL_TIMES = int(os.environ.get("FAIL_TIMES", "2"))
count = 0

class Handler(BaseHTTPRequestHandler):
    def do_GET(self):
        global count
        if self.path == "/flaky":
            count += 1
            if count <= FAIL_TIMES:
                self._send(503, {"error": "temporarily unavailable"})
            else:
                self._send(200, {"items": [{"id": 1, "name": "dummy-a"}, {"id": 2, "name": "dummy-b"}]})
        elif self.path == "/notfound":
            self._send(404, {"error": "not found"})
        else:
            self._send(404, {"error": "unknown path"})

    def _send(self, status, body):
        data = json.dumps(body).encode("utf-8")
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(data)))
        self.end_headers()
        self.wfile.write(data)

    def log_message(self, fmt, *args):
        print("[server]", self.command, self.path, "->", args[1], flush=True)

if __name__ == "__main__":
    HTTPServer(("127.0.0.1", 8000), Handler).serve_forever()

• 1

python3 mock_server.py

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15
• 16
• 17
• 18
• 19
• 20
• 21
• 22
• 23
• 24
• 25
• 26
• 27
• 28
• 29
• 30
• 31
• 32
• 33

"""自分でリトライを書く版：失敗したら 1秒→2秒→4秒 と待ち時間を倍にして再挑戦する。"""
import random
import time

import requests

API_URL = "http://127.0.0.1:8000/flaky"   # 練習用のダミーAPI
RETRY_STATUS = {429, 500, 502, 503, 504}  # 「待てば直るかもしれない」ステータス
MAX_TRIES = 4                             # 最初の1回＋リトライ3回


def fetch_with_retry(url: str) -> dict:
    for attempt in range(1, MAX_TRIES + 1):
        try:
            res = requests.get(url, timeout=10)
        except (requests.ConnectionError, requests.Timeout) as e:
            reason = type(e).__name__
        else:
            if res.status_code not in RETRY_STATUS:
                res.raise_for_status()      # 404 などはリトライせず、すぐエラーにする
                return res.json()
            reason = f"HTTP {res.status_code}"

        if attempt == MAX_TRIES:
            raise RuntimeError(f"{MAX_TRIES}回試しても失敗しました（最後の理由: {reason}）")
        wait = 2 ** (attempt - 1) + random.uniform(0, 0.5)   # 1, 2, 4秒 ＋ ゆらぎ
        print(f"{attempt}回目失敗（{reason}）。{wait:.1f}秒待って再挑戦します")
        time.sleep(wait)


if __name__ == "__main__":
    data = fetch_with_retry(API_URL)
    print("取得できた件数:", len(data["items"]))

• 1
• 2
• 3

1回目失敗（HTTP 503）。1.3秒待って再挑戦します
2回目失敗（HTTP 503）。2.4秒待って再挑戦します
取得できた件数: 2

• 1
• 2
• 3

[server] GET /flaky -> 503
[server] GET /flaky -> 503
[server] GET /flaky -> 200

• 1

RuntimeError: 4回試しても失敗しました（最後の理由: HTTP 503）

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15
• 16
• 17
• 18
• 19
• 20
• 21
• 22
• 23
• 24
• 25
• 26
• 27

"""requests + urllib3 の Retry にリトライを任せる版。"""
import requests
from requests.adapters import HTTPAdapter
from urllib3.util import Retry

API_URL = "http://127.0.0.1:8000/flaky"   # 練習用のダミーAPI


def make_session() -> requests.Session:
    retry = Retry(
        total=3,                                  # リトライは最大3回（最初の1回は含まない）
        backoff_factor=1,                         # 待ち時間の基準（秒）
        status_forcelist=[429, 500, 502, 503, 504],
        allowed_methods=["GET"],                  # 何度送っても結果が変わらない GET だけ
    )
    session = requests.Session()
    adapter = HTTPAdapter(max_retries=retry)
    session.mount("http://", adapter)
    session.mount("https://", adapter)
    return session


if __name__ == "__main__":
    session = make_session()
    res = session.get(API_URL, timeout=10)
    res.raise_for_status()
    print("ステータス:", res.status_code, "件数:", len(res.json()["items"]))

• 1

ステータス: 200 件数: 2

• 1

requests.exceptions.RetryError: HTTPConnectionPool(host='127.0.0.1', port=8000): Max retries exceeded with url: /flaky (Caused by ResponseError('too many 503 error responses'))

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12

次のPythonスクリプトの requests.get の部分に、リトライを入れたいです。

【条件】
- requests.Session に HTTPAdapter と urllib3 の Retry を設定する
- リトライするのは 429, 500, 502, 503, 504 のときだけ。total=3, backoff_factor=1
- リトライするメソッドは GET だけ（POST は入れない）
- timeout=10 を必ず付ける
- 最後に raise_for_status() で失敗を検知する
- APIキーはコードに書かず、環境変数から読む形のままにする

【今のコード】
（ここに貼る。APIキーやURLのトークン部分は伏せる）

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8

Pythonの requests と urllib3 の Retry でAPIを呼んでいます。
次のエラーで止まりました。リトライの設定が正しいか、原因の候補と確認する順番を教えてください。

【エラーメッセージ】
（ここに貼る。URLに含まれるキーやトークン、個人の情報は伏せる）

【Retry の設定】
total=3, backoff_factor=1, status_forcelist=[429, 500, 502, 503, 504], allowed_methods=["GET"]