Tenacityを使ったリトライの紹介

ソフトウェア開発の世界では、一時的な失敗や制御できない問題が原因で、コードを複数回リトライする必要がある状況がよくあります。これには、ネットワークエラー、サーバータイムアウト、APIレート制限などが含まれます。これらの課題に対処するため、Pythonの開発者は強力で柔軟なTenacityライブラリを使用することができます。

Tenacityは、Pythonアプリケーションにリトライロジックを追加するプロセスを簡素化する、広く使われており、メンテナンスが行き届いているライブラリです。簡単に使用できるAPIを提供し、最小限の労力と最大限のカスタマイズでリトライを実装できるようにします。この記事では、Tenacityの多様な機能、基本的な使用方法から高度な戦略までを探り、様々なシナリオで効果的に使用する方法を示します。

Tenacityのインストール

Tenacityの機能に取り組む前に、ライブラリをインストールする必要があります。以下のコマンドを使用して、Pythonパッケージマネージャーであるpipを使用してTenacityをインストールできます。

$ pip install tenacity

Tenacityの基本的な使用方法

シンプルなリトライ

Tenacityを使用するには、tenacityモジュールからretryデコレータをインポートし、リトライする関数に適用します。

from tenacity import retry

@retry
def fetch_data():
    # Code to fetch data from a remote server

デフォルトでは、Tenacityは成功するまで、1秒のウェイトを挟んで、関数を無制限にリトライします。

リトライロジックのカスタマイズ

Tenacityは、様々なオプションを使用してリトライの動作をカスタマイズできます。例えば、リトライ回数、試行間の待機時間、およびリトライが発生する条件を設定できます。以下は、最大5回の試行間に2秒のウェイトを挟んでリトライする関数の例です。

from tenacity import retry, stop_after_attempt, wait_fixed

@retry(stop=stop_after_attempt(5), wait=wait_fixed(2))
def fetch_data():
    # Code to fetch data from a remote server

Tenacityを使った例外処理

リトライを行う場合、例外処理を効果的に行うことが重要です。この章では、Tenacityの例外処理機能、例外を再送出する方法、およびカスタム例外ハンドラを作成する方法について探ります。

例外の再送出

デフォルトでは、Tenacityはリトライ中に発生した例外を抑制します。全ての試行が実行された後でも例外が発生したことを知る必要がある場合、この動作は問題があります。Tenacityを構成して、全ての試行後に最後に発生した例外を再送出するようにするには、reraise=Trueオプションを使用します。

from tenacity import retry, stop_after_attempt

@retry(stop=stop_after_attempt(3), reraise=True)
def fetch_data():
    # Code to fetch data from a remote server

この例では、fetch_data関数が3回の試行後も失敗し続ける場合、Tenacityは最後に発生した例外を再送出し、アプリケーションで適切に処理できるようにします。

カスタム例外ハンドリング

例外が発生した場合に特定のアクションを実行したい場合があります。例えば、エラーをログに記録したり、フォールバック関数を実行したりすることができます。Tenacityは、カスタム例外ハンドリングロジックを定義できるretry_error_callbackというコールバックメカニズムを提供します。

以下は、例外をログに記録し、デフォルト値を返すカスタムエラーコールバックの例です。

from tenacity import retry, stop_after_attempt

def error_callback(retry_state):
    print(f"An error occurred: {retry_state.outcome.exception()}")
    return "default_value"

@retry(stop=stop_after_attempt(3), retry_error_callback=error_callback)
def fetch_data():
    # Code to fetch data from a remote server

この例では、fetch_data関数が3回の試行後も失敗し続ける場合、error_callback関数が呼び出されます。この関数は、例外をログに記録し、デフォルト値を返します。このデフォルト値がfetch_data関数の結果となります。

retry_error_callbackを使用する場合は、Tenacityは例外を再送出しません。例外を処理しつつ再送出するには、エラーコールバック内でそれを行うことができます。

def error_callback(retry_state):
    print(f"An error occurred: {retry_state.outcome.exception()}")
    raise retry_state.outcome.exception()

@retry(stop=stop_after_attempt(3), retry_error_callback=error_callback)
def fetch_data():
    # Code to fetch data from a remote server

参考

https://tenacity.readthedocs.io/en/latest/
https://github.com/jd/tenacity