python-resilience

出典の表記：本スキルの内容は wshobson/agents に基づいています（MIT License、Copyright (c) 2024 Seth Hobson。ライセンス全文は同じディレクトリの LICENSE.txt を参照してください）。 使用、変更、再配布の際は、MIT ライセンスの条項に従い、著作権表示を保持してください。

日本語の説明

本スキルでは、フルスタック開発に関するベストプラクティスとパターンを案内します。

---

name: python-resilience description: 自動リトライ、指数バックオフ、タイムアウト、フォールトトレラントなデコレーターなど、Python のレジリエンスパターンを扱います。リトライロジックの追加、タイムアウトの実装、フォールトトレラントなサービスの構築、一時的な障害への対処に使用してください。

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. リトライ率を監視する - リトライ率が高い場合、根本的な問題があることを示しています

能力の範囲

✅ 得意なこと

• フルスタック開発に関するベストプラクティス
• コードアーキテクチャの設計ガイダンス
• パフォーマンス最適化の提案

⚠️ 素材が必要なこと

• 具体的なプロジェクトのコンテキスト
• 技術スタックの情報

❌ 対象外

• 技術以外の分野におけるビジネス上の意思決定
• 具体的なコードの実装（ガイダンスのみを提供）

ワークフロー

Step 1：要件分析

プロジェクトの要件と技術的な制約を理解します

Step 2：アーキテクチャ設計

システムアーキテクチャとコンポーネントの分割を設計します

Step 3：実装ガイダンス

具体的な実装の提案とコードパターンを提供します

Step 4：コードレビュー

コードの品質とベストプラクティスへの準拠状況を確認します

Step 5：最適化の提案

パフォーマンスの最適化と改善の提案を行います

注意点

1. 過剰設計

単純なケースで複雑な設計パターンを使用しないでください

2. パフォーマンスの軽視

特に本番環境では、パフォーマンスの最適化を軽視しないでください

3. セキュリティへの配慮

常にセキュリティとデータ保護を考慮してください

4. テストカバレッジ

特に境界条件について、十分なテストカバレッジを確保してください

5. ドキュメントのメンテナンス

ドキュメントをコードと同期した状態に保ってください