🔄

Pythonの『tenacity』を使ったお手軽リトライ処理実装

sakairi-y

2024/12/23に公開

1. 概要

プログラムを実行していると、ネットワークエラーなどの要因で一時的に処理が失敗することがあります。
こうしたエラーは多くの場合、一度失敗しても少し時間を置いて再試行（リトライ）すれば成功する可能性があります。
Pythonではライブラリ『tenacity』を利用すれば、デコレートするだけでお手軽にリトライ処理を適用することができます。

2. Tenacityの基本情報

2.1. 各種URL

2.2. ライセンスについて

tenacityはオープンソースであり、Apache License 2.0の下で提供されています。
このライセンスにより、以下が許可されています。

個人および商用での利用
ソースコードの修正と再配布
二次的著作物の作成

※ ただし、再配布の際には元のライセンスを保持する必要があります。

3. 使い方

3.1. 基本の使い方：成功するまでリトライしよう

適用したい関数に対してデコレータ関数「retry」でデコレートするだけです！！
対象の関数にて例外が発生した際にはリトライ処理を実行し、値が返された場合はその値を返します。

import random
from tenacity import retry

@retry
def target_function():
    match random.randint(0, 4):
        case 0:
            print("ValueErrorです！")
            raise ValueError("value error")
        case 1:
            print("TypeErrorです！")
            raise TypeError("type error")
        case 2:
            print("IOErrorです！")
            raise IOError("io error")
        case 3:
            print("戻り値がNoneです！")
            return None
        case _:
            print("成功です！")
            return "success"

target_function()

3.2. 最大リトライ回数とリトライ間隔を設定しよう

大抵の外部APIやデータベースアクセスなどは、過剰なアクセスを許可していません。
負荷をかけないように最大リトライ回数と、リトライを行う間隔を設定して対応しましょう。

最大リトライ数:
- キーワード引数「stop」に「stop_after_attempt」を設定します。
- 設定した回数リトライしても失敗した場合は、例外を投げます。
- キーワード引数「stop」にはリトライの全体時間が設定値を超過したら止める「stop_after_delay」やリトライの全体時間が設定値を超過しそうなら止める「stop_before_delay」なども設定でき、| 演算子を使用して複数の停止条件を組み合わせて利用できます。
リトライ間隔：
- キーワード引数「wait」に「wait_fixed」を設定します。
- 設定した固定待ち時間を待機してからリトライを行います。
- 他にも指数バックオフ指定「wait_exponential」やジッター指定（ランダム）「wait_random」、指数増加ジッター指定「wait_random_exponential」などがあり、それぞれの組み合わせ（固定値＋ジッター）やチェーン構築（n回は固定値、それ以降は指数増加ジッター）も利用できます。

import random
from tenacity import retry, stop_after_attempt, wait_fixed

@retry(
    stop=stop_after_attempt(3), # 最大リトライ数の設定：この場合は3回
    wait=wait_fixed(1), # リトライ間隔の設定：この場合は1秒ごとに実行
)
def target_function():
    match random.randint(0, 4):
        case 0:
            print("ValueErrorです！")
            raise ValueError("value error")
        case 1:
            print("TypeErrorです！")
            raise TypeError("type error")
        case 2:
            print("IOErrorです！")
            raise IOError("io error")
        case 3:
            print("戻り値がNoneです！")
            return None
        case _:
            print("成功です！")
            return "success"

target_function()

3.3. 特定の例外や戻り値の場合にリトライさせよう

特定のエラー（例：TimeoutErrorやConnectionErrorなど）についてはリトライがしたいが、それ以外はエラーとして処理させたいことありますよね。
また、エラーじゃないけど戻り値が特定の値（例：APIでNoneが返ってきたなど）でリトライしたいこともありますよね。
そんな場合には以下のように設定することで対応できます。

特定のエラーだけリトライ:
- キーワード引数「retry」に「retry_if_exception_type」を設定します。
- 設定したエラーの場合はリトライし、それ以外はそのまま例外を投げます。
戻り値に基づくリトライ：
- キーワード引数「retry」に「retry_if_result」を設定し、条件を関数で自由に指定できます。
- 設定した関数の判定がTrueの場合、リトライを行います。

他にも特定エラー以外の場合リトライする「retry_if_not_exception_type」や例外メッセージを条件とする「retry_if_exception_message」などがあり、複数の組み合わせ（and条件やor条件）も利用できます。

キーワード引数「`retry`」に設定できるオプション一覧

オプション名	条件	動作
`retry_if_exception`	任意の例外	条件関数がTrueで再試行
`retry_if_exception_type`	指定された例外タイプ	一致する例外で再試行
`retry_if_not_exception_type`	指定された例外タイプ以外	一致しない例外で再試行
`retry_unless_exception_type`	指定された例外タイプが発生しない場合	再試行
`retry_if_result`	関数の結果	条件関数がTrueで再試行
`retry_if_not_result`	関数の結果	条件関数がFalseで再試行
`retry_if_exception_message`	例外メッセージ	条件関数がTrueで再試行
`retry_if_not_exception_message`	例外メッセージ	条件関数がFalseで再試行
`retry_any`	複数の条件	いずれかがTrueで再試行
`retry_all`	複数の条件	すべてがTrueで再試行

import random
from tenacity import retry, retry_any, retry_if_exception_type, retry_if_result

@retry(
    retry=retry_any(
        retry_if_exception_type(IOError), # 特定エラーだけリトライ：この場合はIOError
        retry_if_result(lambda result: result is None) # 戻り値に基づくリトライ：この場合はNoneの時
    )
)
def target_function():
    match random.randint(0, 4):
        case 0:
            print("ValueErrorです！")
            raise ValueError("value error")
        case 1:
            print("TypeErrorです！")
            raise TypeError("type error")
        case 2:
            print("IOErrorです！")
            raise IOError("io error")
        case 3:
            print("戻り値がNoneです！")
            return None
        case _:
            print("成功です！")
            return "success"

target_function()

3.4. リトライの前後に処理を追加しよう

以下の設定にてリトライの前後に、状況を監視したりなどの自由な処理が追加可能です。

リトライ前に処理を追加:
- キーワード引数「before」に追加する処理の関数を設定します。
- 設定したエラーの場合はリトライし、それ以外はそのまま例外を投げます。
リトライ後に処理を追加：
- キーワード引数「after」に追加する処理の関数を設定します。
- 設定した関数の判定がTrueの場合、リトライを行います。

import random
from tenacity import retry, RetryCallState

def before_retry(retry_state: RetryCallState):
    # RetryCallState: 現在のリトライ状態（例：試行回数、最後のエラーなど）を提供するオブジェクトです。
    print(f"リトライ開始（{retry_state.attempt_number}回目）")

def after_retry(retry_state: RetryCallState):
    print(f"リトライ終了（{retry_state.attempt_number}回目）処理時間：{retry_state.outcome_timestamp - retry_state.start_time}")

@retry(
    before=before_retry,
    after=after_retry,
)
def target_function():
    match random.randint(0, 4):
        case 0:
            print("ValueErrorです！")
            raise ValueError("value error")
        case 1:
            print("TypeErrorです！")
            raise TypeError("type error")
        case 2:
            print("IOErrorです！")
            raise IOError("io error")
        case 3:
            print("戻り値がNoneです！")
            return None
        case _:
            print("成功です！")
            return "success"

target_function()

4. まとめ

tenacityは、柔軟なリトライ機能を簡単に実現できるので非常におすすめです。
公式ドキュメントもわかりやすいので、ぜひ利用してみてください。

Aidemy Tech BlogPublication

株式会社アイデミーの技術ブログです。