信頼できない環境で稼働するアプリケーションには、リトライ処理が不可欠です。
難しくはないので自前で実装してしまうのですが、特定の例外のみリトライしたい・リトライ間隔を指数関数的に増やしたい・リトライ時はログ出力したいなどの細かなリクエストをそれぞれ記述してしまうと、本質的な処理が追いづらくなるので、可能であればライブラリを使って解決したくなります。
tenacity
リトライを簡単に実装するためのPythonライブラリにもいくつかあるのですが、今回は最近でもアップデートされている tenacity を紹介します。類似ライブラリとしてretryやretryingが挙げられますが、いずれも数年以上リリースされていません。
pip install tenacityでインストールできます。以降のコードはこちらのバージョンで実行してます。
$ python --version
Python 3.8.6
$ pip list | grep tenacity
tenacity 6.2.0
retryを関数 (ここでは一定確率で失敗するfail_sometimesを定義) にアノテーションするだけでリトライ処理が実装できます。引数無しですと、成功するまで間隔を開けずに繰り返されます。
import random
import time
from tenacity import retry
@retry
def fail_sometimes(n: int):
v = random.choice(list(range(n)))
time.sleep(1)
print("v:", v)
if v > 0:
raise IOError("Error")
fail_sometimes(10)
停止条件
retry関数はRetryingのラッパーなので、Retryingのコンストラクタを見ると設定可能なパラメータがわかります。
例えばstopには、リトライする回数 (stop_after_attempt) や制限時間 (stop_after_delay) を設けることができます。
from tenacity import retry, stop_after_attempt
@retry(stop=stop_after_attempt(3))
def fail_sometimes(n: int):
fail_sometimes(10)
stop_after_delayはリトライ時に指定した秒数を経過していなければ処理開始します。処理の途中で打ち切られることはありません。
from tenacity import retry, stop_after_delay
@retry(stop=stop_after_delay(3))
def fail_sometimes(n: int):
v = random.choice(list(range(n)))
time.sleep(2)
print("v:", v)
if v > 0:
raise IOError("Error")
fail_sometimes(10)
リトライ間隔
リトライ間隔はwaitパラメータで設定します。
一定時間の場合はwait_fixed、ランダム時間の場合はwait_random、指数関数で増やす場合はwait_exponentialをそれぞれ使います。
import random
import time
from datetime import datetime
from tenacity import retry, wait_fixed
@retry(wait=wait_fixed(2))
def fail_sometimes(n: int):
v = random.choice(list(range(n)))
print(datetime.now(), "v:", v)
if v > 0:
raise IOError("Error")
fail_sometimes(5)
リトライ条件
特定の例外のみリトライして、それ以外はそのままraiseしてほしいケースがあります。retryパラメータにretry_if_exception_typeを渡すことで実現できます。
以下の場合、IOErrorのみリトライして、それ以外の例外はraiseされて終了します。
from tenacity import retry, retry_if_exception_type
@retry(retry=retry_if_exception_type(IOError))
def fail_sometimes(n: int):
v = random.choice(list(range(n)))
print(datetime.now(), "v:", v)
if v == 1:
raise ValueError("Value Error")
elif v > 1:
raise IOError("IO Error")
fail_sometimes(5)
ログ出力
リトライ時にログ出力もbeforeやafterのパラメータで設定できます。以下ではafter_logを使ってリトライ時にログ出力するようにしています。
import logging
from tenacity import retry, after_log
logger = logging.getLogger(__name__)
@retry(after=after_log(logger, logging.WARNING))
def fail_sometimes(n: int):
v = random.choice(list(range(n)))
print(datetime.now(), "v:", v)
if v > 0:
raise IOError("IO Error")
fail_sometimes(5)
