Python Requestsのリトライ設定：再試行回数・待機時間・プロキシエラーを制御する

python3 -m venv .venv
source .venv/bin/activate
python -m pip install requests==2.34.2 urllib3==2.7.0

py -m venv .venv
.\.venv\Scripts\python.exe -m pip install requests==2.34.2 urllib3==2.7.0

import requests
from requests.adapters import HTTPAdapter
from urllib3.util import Retry


def build_session() -> requests.Session:
    retry = Retry(
        total=4,
        connect=4,
        read=4,
        status=4,
        other=0,
        allowed_methods=frozenset({"GET", "HEAD", "OPTIONS"}),
        status_forcelist={429, 500, 502, 503, 504},
        backoff_factor=0.25,
        backoff_jitter=0.1,
        respect_retry_after_header=True,
        raise_on_status=False,
    )

    adapter = HTTPAdapter(
        max_retries=retry,
        pool_connections=10,
        pool_maxsize=10,
    )
    session = requests.Session()
    session.trust_env = False
    session.mount("http://", adapter)
    session.mount("https://", adapter)
    return session


if __name__ == "__main__":
    import sys
    path = sys.argv[1] if len(sys.argv) > 1 else "/flaky"
    if path not in {"/flaky", "/rate-limit", "/missing"}:
        raise SystemExit("/flaky、/rate-limit、/missingのいずれかを指定してください")
    with build_session() as session:
        response = session.get("http://127.0.0.1:8877" + path, timeout=(5, 20))
        print("再試行履歴:", [h.status for h in response.raw.retries.history])
        print("最終ステータス:", response.status_code)
        response.raise_for_status()
        print("JSON:", response.json())

from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
import json
from threading import Lock

COUNTS = {"flaky": 0, "rate-limit": 0}
LOCK = Lock()


class Handler(BaseHTTPRequestHandler):
    def send_json(self, status, payload, headers=None):
        body = json.dumps(payload).encode("utf-8")
        self.send_response(status)
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("Content-Length", str(len(body)))
        for name, value in (headers or {}).items():
            self.send_header(name, value)
        self.end_headers()
        self.wfile.write(body)

    def do_GET(self):
        if self.path == "/reset":
            with LOCK:
                COUNTS.update({"flaky": 0, "rate-limit": 0})
            self.send_json(200, {"ok": True, "message": "counters reset"})
            return

        if self.path == "/flaky":
            with LOCK:
                COUNTS["flaky"] += 1
                attempt = COUNTS["flaky"]
            status = 503 if attempt <= 2 else 200
            self.send_json(status, {"ok": status == 200, "attempt": attempt})
            return

        if self.path == "/rate-limit":
            with LOCK:
                COUNTS["rate-limit"] += 1
                attempt = COUNTS["rate-limit"]
            if attempt == 1:
                self.send_json(
                    429,
                    {"ok": False, "attempt": attempt},
                    {"Retry-After": "1"},
                )
            else:
                self.send_json(200, {"ok": True, "attempt": attempt})
            return

        self.send_json(404, {"ok": False, "message": "not found"})

    def log_message(self, fmt, *args):
        print(f"サーバー {self.command} {self.path} -> {args[1]}", flush=True)


if __name__ == "__main__":
    server = ThreadingHTTPServer(("127.0.0.1", 8877), Handler)
    print("待ち受け開始: http://127.0.0.1:8877", flush=True)
    server.serve_forever()

python flaky_server.py

curl --fail --show-error http://127.0.0.1:8877/reset
python httpadapter_demo.py /flaky

curl --fail --show-error http://127.0.0.1:8877/reset
python httpadapter_demo.py /rate-limit
python httpadapter_demo.py /missing

import random
import time
import requests

RETRYABLE_STATUSES = {500, 502, 503, 504}


def _get_with_basic_retry(session, url: str, attempts: int) -> requests.Response:
    if attempts < 1:
        raise ValueError("attemptsには1以上を指定してください")

    last_error = None
    for attempt in range(1, attempts + 1):
        try:
            response = session.get(url, timeout=(5, 20), allow_redirects=False)
            if response.status_code not in RETRYABLE_STATUSES:
                response.raise_for_status()
                return response
            response.close()
            last_error = requests.HTTPError(
                f"retryable HTTP {response.status_code}", response=response
            )
        except requests.exceptions.SSLError:
            raise
        except requests.exceptions.ProxyError:
            raise
        except (
            requests.exceptions.ConnectTimeout,
            requests.exceptions.ReadTimeout,
            requests.exceptions.ConnectionError,
        ) as exc:
            last_error = exc

        if attempt < attempts:
            delay = 0.5 * (2 ** (attempt - 1)) + random.uniform(0, 0.1)
            time.sleep(delay)

    raise RuntimeError(f"{attempts}回の試行でリクエストに失敗しました") from last_error

def get_with_basic_retry(url: str, attempts: int = 3) -> requests.Response:
    with requests.Session() as session:
        session.trust_env = False
        return _get_with_basic_retry(session, url, attempts)


if __name__ == "__main__":
    response = get_with_basic_retry("http://127.0.0.1:8877/flaky")
    print("最終ステータス:", response.status_code)
    print("JSON:", response.json())

python basic_retry.py

from datetime import datetime, timezone
from email.utils import parsedate_to_datetime


def retry_after_seconds(response):
    value = response.headers.get("Retry-After")
    if not value:
        return None
    value = value.strip()
    if value.isascii() and value.isdigit():
        return max(0.0, float(value))

    try:
        retry_at = parsedate_to_datetime(value)
    except (TypeError, ValueError, OverflowError):
        return None

    if retry_at.tzinfo is None:
        retry_at = retry_at.replace(tzinfo=timezone.utc)
    return max(0.0, (retry_at - datetime.now(timezone.utc)).total_seconds())

export ROLA_PROXY_HOST="proxy.example"
export ROLA_PROXY_PORT="1000"
export ROLA_PROXY_USERNAME="管理画面で生成した完全なユーザー名"
export ROLA_PROXY_PASSWORD="管理画面のパスワード"

$env:ROLA_PROXY_HOST="proxy.example"
$env:ROLA_PROXY_PORT="1000"
$env:ROLA_PROXY_USERNAME="管理画面で生成した完全なユーザー名"
$env:ROLA_PROXY_PASSWORD="管理画面のパスワード"

import os
from urllib.parse import quote
import requests
from requests.adapters import HTTPAdapter


def build_proxy_url():
    names = ("ROLA_PROXY_HOST", "ROLA_PROXY_PORT",
             "ROLA_PROXY_USERNAME", "ROLA_PROXY_PASSWORD")
    if any(not os.environ.get(name) for name in names):
        raise ValueError("必要なプロキシ環境変数が未設定です")
    host = os.environ["ROLA_PROXY_HOST"]
    port = int(os.environ["ROLA_PROXY_PORT"])
    if not 1 <= port <= 65535 or any(c in host for c in "/@?# "):
        raise ValueError("ホストまたはポートを確認してください")
    username = quote(os.environ["ROLA_PROXY_USERNAME"], safe="")
    password = quote(os.environ["ROLA_PROXY_PASSWORD"], safe="")
    return f"http://{username}:{password}@{host}:{port}"


def main():
    try:
        proxy_url = build_proxy_url()
        with requests.Session() as session:
            session.trust_env = False
            session.mount("https://", HTTPAdapter(max_retries=0))
            response = session.get(
                "https://api64.ipify.org?format=json",
                proxies={"http": proxy_url, "https": proxy_url},
                timeout=(10, 30),
                allow_redirects=False,
            )
            if response.status_code != 200:
                raise SystemExit(f"停止: HTTP {response.status_code}")
            if "application/json" not in response.headers.get("Content-Type", ""):
                raise ValueError("JSON以外の応答です")
            data = response.json()
            import ipaddress
            ipaddress.ip_address(data["ip"])
            print("出口確認サービスへの接続に成功しました。IPはログに出力していません。")
    except requests.exceptions.ProxyError:
        raise SystemExit("停止: プロキシ接続または認証を確認してください") from None
    except requests.exceptions.SSLError:
        raise SystemExit("停止: TLS証明書とCA設定を確認してください") from None
    except requests.exceptions.Timeout:
        raise SystemExit("停止: 通信がタイムアウトしました") from None
    except requests.RequestException:
        raise SystemExit("停止: 通信エラーが発生しました") from None
    except (ValueError, KeyError, TypeError):
        raise SystemExit("停止: 環境変数または応答データを確認してください") from None


if __name__ == "__main__":
    main()

python proxy_check.py

3 ×（初回1回＋再試行4回）＝最大15試行
50ジョブ × 15試行＝累計最大750試行

custom_settings = {
    "RETRY_ENABLED": True,
    "RETRY_TIMES": 2,
    "RETRY_HTTP_CODES": [429, 500, 502, 503, 504],
}