python-resilience

Python レジリエンスパターン

一時的な障害、ネットワークの問題、サービスの停止を適切に処理する耐障害性の高い Python アプリケーションを構築します。レジリエンスパターンは、依存関係が信頼できない場合でもシステムを稼働させ続けます。

このスキルを使用する場面

• 外部サービス呼び出しにリトライロジックを追加するとき
• ネットワーク操作にタイムアウトを実装するとき
• 耐障害性の高いマイクロサービスを構築するとき
• レート制限とバックプレッシャーを処理するとき
• インフラストラクチャのデコレーターを作成するとき
• サーキットブレーカーを設計するとき

中核となる概念

1. 一時的な障害と恒久的な障害

一時的なエラー（ネットワークタイムアウト、一時的なサービス問題）はリトライします。恒久的なエラー（無効な認証情報、不正なリクエスト）はリトライしません。

2. 指数バックオフ

復旧中のサービスに負荷をかけすぎないよう、リトライ間の待機時間を増やします。

3. ジッター

多数のクライアントが同時にリトライする際のサンダリングハードを防ぐため、バックオフにランダム性を加えます。

4. 上限付きリトライ

試行回数と合計時間の両方に上限を設け、無限のリトライループを防ぎます。

クイックスタート

from tenacity import retry, stop_after_attempt, wait_exponential_jitter

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential_jitter(initial=1, max=10),
)
def call_external_service(request: dict) -> dict:
    return httpx.post("https://api.example.com", json=request).json()

基本的なパターン

パターン 1: Tenacity による基本的なリトライ

本番環境で使えるリトライロジックには tenacity ライブラリを使用します。より単純なケースでは、組み込みのリトライ機能や軽量なカスタム実装を検討してください。

from tenacity import (
    retry,
    stop_after_attempt,
    stop_after_delay,
    wait_exponential_jitter,
    retry_if_exception_type,
)

TRANSIENT_ERRORS = (ConnectionError, TimeoutError, OSError)

@retry(
    retry=retry_if_exception_type(TRANSIENT_ERRORS),
    stop=stop_after_attempt(5) | stop_after_delay(60),
    wait=wait_exponential_jitter(initial=1, max=30),
)
def fetch_data(url: str) -> dict:
    """Fetch data with automatic retry on transient failures."""
    response = httpx.get(url, timeout=30)
    response.raise_for_status()
    return response.json()

パターン 2: 適切なエラーのみをリトライする

特定の一時的な例外をホワイトリストに登録します。以下は決してリトライしません:

• ValueError、TypeError - これらはバグであり、一時的な問題ではありません
• AuthenticationError - 無効な認証情報が有効になることはありません
• HTTP 4xx エラー（429 を除く） - クライアントエラーは恒久的です

from tenacity import retry, retry_if_exception_type
import httpx

# Define what's retryable
RETRYABLE_EXCEPTIONS = (
    ConnectionError,
    TimeoutError,
    httpx.ConnectTimeout,
    httpx.ReadTimeout,
)

@retry(
    retry=retry_if_exception_type(RETRYABLE_EXCEPTIONS),
    stop=stop_after_attempt(3),
    wait=wait_exponential_jitter(initial=1, max=10),
)
def resilient_api_call(endpoint: str) -> dict:
    """Make API call with retry on network issues."""
    return httpx.get(endpoint, timeout=10).json()

パターン 3: HTTP ステータスコードのリトライ

一時的な問題を示す特定の HTTP ステータスコードをリトライします。

from tenacity import retry, retry_if_result, stop_after_attempt
import httpx

RETRY_STATUS_CODES = {429, 502, 503, 504}

def should_retry_response(response: httpx.Response) -> bool:
    """Check if response indicates a retryable error."""
    return response.status_code in RETRY_STATUS_CODES

@retry(
    retry=retry_if_result(should_retry_response),
    stop=stop_after_attempt(3),
    wait=wait_exponential_jitter(initial=1, max=10),
)
def http_request(method: str, url: str, **kwargs) -> httpx.Response:
    """Make HTTP request with retry on transient status codes."""
    return httpx.request(method, url, timeout=30, **kwargs)

パターン 4: 例外とステータスの複合リトライ

ネットワーク例外と HTTP ステータスコードの両方を処理します。

from tenacity import (
    retry,
    retry_if_exception_type,
    retry_if_result,
    stop_after_attempt,
    wait_exponential_jitter,
    before_sleep_log,
)
import logging
import httpx

logger = logging.getLogger(__name__)

TRANSIENT_EXCEPTIONS = (
    ConnectionError,
    TimeoutError,
    httpx.ConnectError,
    httpx.ReadTimeout,
)
RETRY_STATUS_CODES = {429, 500, 502, 503, 504}

def is_retryable_response(response: httpx.Response) -> bool:
    return response.status_code in RETRY_STATUS_CODES

@retry(
    retry=(
        retry_if_exception_type(TRANSIENT_EXCEPTIONS) |
        retry_if_result(is_retryable_response)
    ),
    stop=stop_after_attempt(5),
    wait=wait_exponential_jitter(initial=1, max=30),
    before_sleep=before_sleep_log(logger, logging.WARNING),
)
def robust_http_call(
    method: str,
    url: str,
    **kwargs,
) -> httpx.Response:
    """HTTP call with comprehensive retry handling."""
    return httpx.request(method, url, timeout=30, **kwargs)

詳細な実例とパターン

詳細なセクション（## Advanced Patterns で始まるもの）は references/details.md にあります。上記のナビゲーション概要では不十分な場合、そのファイルを参照してください。

ベストプラクティスのまとめ

1. 一時的なエラーのみをリトライする - バグや認証失敗はリトライしない
2. 指数バックオフを使用する - サービスが復旧する時間を与える
3. ジッターを追加する - 同期したリトライによるサンダリングハードを防ぐ
4. 合計時間に上限を設ける - stop_after_attempt(5) | stop_after_delay(60)
5. すべてのリトライをログに記録する - サイレントなリトライはシステム的な問題を隠します
6. デコレーターを使用する - リトライロジックをビジネスロジックから分離する
7. 依存関係を注入する - インフラストラクチャをテスト可能にする
8. どこでもタイムアウトを設定する - すべてのネットワーク呼び出しにタイムアウトが必要です
9. 適切に失敗する - 重要でないパスにはキャッシュされた値やデフォルト値を返す
10. リトライ率を監視する - 高いリトライ率は根本的な問題を示します