信頼できない環境で稼働するアプリケーションには、リトライ処理が不可欠です。
難しくはないので自前で実装してしまうのですが、特定の例外のみリトライしたい・リトライ間隔を指数関数的に増やしたい・リトライ時はログ出力したいなどの細かなリクエストをそれぞれ記述してしまうと、本質的な処理が追いづらくなるので、可能であればライブラリを使って解決したくなります。
tenacity
リトライを簡単に実装するためのPythonライブラリにもいくつかあるのですが、今回は最近でもアップデートされている tenacity を紹介します。類似ライブラリとしてretryやretryingが挙げられますが、いずれも数年以上リリースされていません。
- PyPI: https://pypi.org/project/tenacity/
- GitHub: https://github.com/jd/tenacity
- Document: https://tenacity.readthedocs.io/en/latest/
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)) # 合計3回まで def fail_sometimes(n: int): # ... fail_sometimes(10) # 実行すると3回まで試して、それでもダメならIOError # $ python tenacity_example/main.py # v: 6 # v: 4 # v: 2 # Traceback (most recent call last): # ... # IOError: Error
stop_after_delayはリトライ時に指定した秒数を経過していなければ処理開始します。処理の途中で打ち切られることはありません。
from tenacity import retry, stop_after_delay @retry(stop=stop_after_delay(3)) # 開始から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) # $ python tenacity_example/main.py # v: 9 # v: 5 # Traceback (most recent call last): # ... # IOError: Error
リトライ間隔
リトライ間隔は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)) # 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) # $ python tenacity_example/main.py # 2020-11-21 15:20:34.616393 v: 1 # 2020-11-21 15:20:36.618325 v: 4 # 2020-11-21 15:20:38.623362 v: 0
リトライ条件
特定の例外のみリトライして、それ以外はそのまま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) # $ python tenacity_example/main.py # 2020-11-21 15:34:06.285533 v: 2 # 2020-11-21 15:34:06.285752 v: 1 # Traceback (most recent call last): # ... # ValueError: Value Error
ログ出力
リトライ時にログ出力も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) # $ python tenacity_example/main.py # 2020-11-21 15:53:27.852080 v: 4 # Finished call to '__main__.fail_sometimes' after 0.000(s), this was the 1st time calling it. # 2020-11-21 15:53:27.852484 v: 0