[Python] tenacity ライブラリを利用してリトライ処理を簡単に実装する

Qiitaにも投稿中: https://qiita.com/engineer-taro/items/39cec5e6119c89c8bb32

Pythonでリトライ処理を書く際に導入すべきライブラリは？

Pythonでリトライ処理を簡単に導入するためのライブラリを検索すると、以下の3つが検索に上がってきます。

• tenacity
• retry
• retrying

今回は__tenacity__についての記事です。

ちなみに、tenacityは「粘り強い」みたいな意味だそうです。

retryとretryingではダメなの？

簡単に理由を表で説明すると以下になります。

ライブラリ	最終更新	スター数
tenacity	記事を書いている前日	3300
retry	2016年5月	420
retrying	2016年7月	1700
※表は本記事を書いている2021年10月5日時点での情報

日本語で検索するとretryとretryingを使っている技術記事が多く出てくるのですが、これらのライブラリは更新が随分昔で止まっています。

retry 最終更新: 2016年5月
retrying 最終更新: 2016年7月

一方で、tenacityは更新され続けており、現在この記事を書いている前日(2021年10月4日)にも更新が入っています。

githubのスター数を見ても、tenacityが圧倒的に多くなっています。
※2021年10月時点

• retry スター数: 420
• retrying スター数: 1700
• tenacity スター数: 3300

これらの理由から、今後も使い続けられるtenacityをリトライのライブラリに選ぶのが良いでしょう。

またtenacity自体、更新の止まったretryingライブラリからフォークして作られています。
そういった意味でもtenacityは最新版のretryingライブラリと言えるでしょう。

tenacityの使い方

基本はgithubのREADMEを参考にするので問題ないですが、細かく使い方を見ていきましょう。
全て紹介するのはボリューム的に難しいので、よく使うものだけ紹介していきます。かなり細かい動作が指定できるので、細かいリトライの制御を必要とする場合はgithubのREADMEを参考にしてください。
※使う際は、pip installをしてください。

pip install tenacity

基本：関数にリトライ処理を導入する

最も基本の使い方は、特定の関数にデコレータとして使用することです。
デコレータを指定すると、例外が発生した場合はリトライ処理が行われることになります。

サンプルコード

sample.py

import random
from tenacity import retry


@retry
def retry_test():
    if random.randint(0, 10) > 1:
        print('retry')
        raise Exception
    else:
        return "Success!"


print(retry_test())

実行結果

retry
retry
retry
retry
Success!

リトライ回数を指定する

リトライ回数を指定する際は、@retryデコレータの引数stopに値を渡す必要があります。
また、その際にtenacityモジュールからstop_after_attemptをインポートする必要があることも忘れないようにしましょう。

サンプルコード

sample.py

from tenacity import retry, stop_after_attempt  # 新たにimportを行う必要がある


@retry(stop=stop_after_attempt(3))
def stop_after_attempt():
    print('retry')
    raise Exception('retry error')


stop_after_attempt()

実行結果
3回実行されたあと、エラーで落ちる。

retry
retry
retry
Exception: retry error

一定時間リトライし続ける

リトライ期間を指定する際は、@retryデコレータの引数stopに値を渡す必要があります。
また、その際にtenacityモジュールからstop_after_delayをインポートする必要があることも忘れないようにしましょう。

※ここで紹介しているのは__一定時間__リトライし続ける実装であり、__一定時間ごと__にリトライを行う実装ではないことに注意してください。

sample.py

from tenacity import retry, stop_after_delay


@retry(stop=stop_after_delay(10))
def stop_after_delay_test():
    print('retry')
    raise Exception


stop_after_delay_test()

実行結果

"""
retry
retry
=======retry出力続く
retry
10秒後に止まる
"""

リトライ回数制限 & 時間制限

回数制限と時間制限をどちらもかけることができます。
下記はどちらかの制限にかかった場合、リトライを停止するサンプルコードです。

サンプルコード

sample.py

import time
from tenacity import retry, stop_after_attempt, stop_after_delay


@retry(stop=(stop_after_attempt(3) | stop_after_delay(10)))
def stop_after_attempt_delay_combine(sleep: int):
    time.sleep(sleep)
    print('retry')
    raise Exception('retry error')


stop_after_attempt_delay_combine(1)  # => 1秒ごとにリトライされるので試行回数の制限でリトライが終了する
stop_after_attempt_delay_combine(5)  # => 5秒ごとにリトライされるので時間制限でリトライが終了する

時間をおいてリトライ

APIアクセスのリトライなど、短期間でアクセスしすぎるとサービス提供側から危険なIPと判断されて弾かれてしまう場合があります。そんなときは、一定時間ごとのリトライを利用しましょう。
秒数固定で一定時間ごとのリトライをさせたい場合、wait引数にwait_fixed(秒)をわたしてあげます。

サンプルコード

sample.py

@retry(wait=wait_fixed(2))
def wait_fixed_test():
    print('retry')
    raise Exception


wait_fixed_test()

実行結果

他には以下のような実装が可能です。

• 指定した範囲の秒数でリトライ処理を行うwait_random(min=秒, max=秒)
• 指数関数的にリトライ待機時間を増やすwait_exponential(multiplier=倍率, min=秒, max=秒)⇒こちらは後で説明
• 特定回数までは3秒ごとにリトライ、特定回数からは5秒ごとにリトライなど、細かい設定

細かい実装を実現したいときは、githubのREADMEを見てください。

指数関数のリトライは個別に解説します。

指数関数的にリトライ待機時間を設定する

指数関数的にリトライ待機時間を増やしたい場合、wait_exponential(multiplier=1, min=3, max=50)を利用します。
multiplierで指定した数字は「[指定した数値]*2」を表す数値です。
分かりにくいと思うので実例を見ていきましょう。

multiplierに1を設定すると、以下の動きになります。

• 1回目のリトライ 1 = 1秒
• 2回目のリトライ 1*2 = 2秒
• 3回目のリトライ 2*2 = 4秒
• 4回目のリトライ 4*2 = 16秒
• 5回目のリトライ 16*2 = 32秒

multiplierに3を設定すると、以下の動きになります。

• 1回目のリトライ 3 = 3秒
• 2回目のリトライ 3*2 = 6秒
• 3回目のリトライ 6*2 = 12秒
• 4回目のリトライ 12*2 = 24秒
• 5回目のリトライ 24*2 = 48秒

minは最低でも待機する時間を設定します。なので、先ほどの1回目のリトライで3秒のとき、min=5で指定されていた場合、5秒の待機を行います。
maxはその逆で、先ほどの5回目のリトライで48秒のとき、max=40で指定されていた場合、40秒の待機を行います。

サンプルコード

sample.py

from time import time
from tenacity import retry, wait_exponential

t = time()


@retry(wait=wait_exponential(multiplier=1, min=3, max=50))
def wait_exponential_sample():
    c = time()
    print(int(c-t))
    print('retry')
    raise Exception

実行結果

0
retry
3
retry
6
retry
10
retry
18
retry

特定のエラー発生時、リトライする

特定のエラーを指定することで、そのエラーの発生時のみリトライを行うことができます。
retry_if_exception_typeをインポートし、リトライさせたい例外を渡します。
特定のエラーを指定できるのは、AWSから返ってきた例外を指定したりAPIから返ってきたエラーを指定できるので、実用度が高いです。

@retry(retry=retry_if_exception_type(TypeError))

複数のエラーをしていするには、タプルの形式でエラーを渡してあげましょう。
@retry(retry=retry_if_exception_type((TypeError, KeyError)))

全体のサンプルコードは以下です。

リトライするサンプルコード

sample.py

from tenacity import retry, retry_if_exception_type


@retry(retry=retry_if_exception_type((TypeError, IndexError)))
def retry_if_exception_sample():
    print('retry')
    raise TypeError # 指定した例外なのでリトライする


retry_if_exception_sample()

リトライしないサンプルコード

sample.py

from tenacity import retry, retry_if_exception_type


@retry(retry=retry_if_exception_type((TypeError, IndexError)))
def retry_if_exception_sample():
    print('retry')
    raise KeyError　# 指定した例外ではないのでリトライしない


retry_if_exception_sample()

また、コード内でリトライさせたい際に、TryAgain例外を投げることも可能です。
分かりにくいと思うのでまたサンプルコードを見ていきましょう。

sample.py

from tenacity import retry, TryAgain # TryAgainをインポートする


@retry
def retry_if_exception_sample(num):
    if num > 10:
        print('retry')
        raise TryAgain
    else:
        return 'Success'


retry_if_exception_sample(11)

TryAgainを使えばどこでも例外を投げられるので、コーディングの自由度が増しそうですね。
とはいえ、使いすぎると汚いコードになりそうです。

特定の返り値の時、リトライする

特定の返り値の時、リトライする処理を書くこともできます。
retry_if_resultをインポートし、返り値を判定する関数を渡してあげましょう。
この説明だけでは分かりにくいので、サンプルコードを見ていきます。

サンプルコード

sample.py

from tenacity import retry, retry_if_result


def check_result(result):
    """返り値がFailedならTrueを返す"""
    return result == 'Failed'


@retry(retry=retry_if_result(check_result))  # 関数をretry_if_resultに渡す
def retry_if_result_sample(num):
    print('retry')
    if num < 10:
        return 'Failed'
    return 'Success'


print(retry_if_result_sample(5))

返り値によってリトライの判断を出来るのも、使い勝手が良いです。
チェック用の関数の中でDBから値を取ってきて比較のような動きも出来るので、状況に応じた動的なリトライ処理も可能になります。

リトライ前後にログを仕込む

以下のように指定することで、前後どちらか、前後両方にログを仕込むことができます。

@retry(before=before_log(logger, logging.DEBUG))

実行結果と一緒にサンプルコードを見ていきましょう。

サンプルコード

sample.py

from tenacity import retry, wait_fixed, before_log, after_log
import sys
import logging

logging.basicConfig(stream=sys.stderr, level=logging.DEBUG)
logger = logging.getLogger(__name__)


@retry(wait=wait_fixed(2),
       before=before_log(logger, logging.DEBUG),
       after=after_log(logger, logging.DEBUG))
def wait_fixed_sample():
    print('retry')
    raise Exception


wait_fixed_sample()

実行結果

DEBUG:__main__:Starting call to '__main__.wait_fixed_sample', this is the 1st time calling it.
retry
DEBUG:__main__:Finished call to '__main__.wait_fixed_sample' after 0.000(s), this was the 1st time calling it.
DEBUG:__main__:Starting call to '__main__.wait_fixed_sample', this is the 2nd time calling it.
retry
DEBUG:__main__:Finished call to '__main__.wait_fixed_sample' after 2.015(s), this was the 2nd time calling it.

実行回数や、リトライ初回からの経過秒数がログ出力されます。
リトライ周りで何か不具合が生じたときのために、ログ出力をしておくとよいでしょう。

その他の機能

このほかにも様々な便利機能があるのですが、基本的にgithubのREADMEに書かれています。
ここまで読まれた方ならREADMEの読み方も分かるはずなので、そのほかの機能は__「こんな機能もあるよ！」__という紹介までにとどめておこうと思います。
（これ以上書くとキリがないと思ってしまった...）

• リトライ前とリトライ後にログを仕込む（Logging）
• リトライの統計情報を表示する（Statistics）
• リトライの制限まで達したあと、特定の関数を実行し返り値を返す（Callback）
• 関数呼び出し時にリトライ対象として指定する
• 関数ではなく特定のコードブロックをリトライ対象とする

サンプルコード

今回解説であげたサンプルコードはgithubで公開しておきます。
圧倒的に公式が分かりやすいので、公式のgithubリポジトリを見て分からない場合のみ、お使いください。

https://github.com/engineer-taro/tenacity_sample