tenacityについて
プログラムを書いていて、HTTPの通信などでリトライ処理を実装する機会は多いと思います。
今回はそんなリトライ処理を簡潔に書けるtenacityの使い方を説明します。
インストール
インストールはpipで可能です。
インストール
pip install tenacity
使い方
シンプルな例
シンプルな例
import random
from tenacity import retry
@retry
def random_error():
num = random.randint(0, 10)
if num > 4:
print(f"Error: num={num}")
raise Exception("Error!")
else:
return f"Success: num={num}"
print(random_error())
# =>
# Error: num=9
# Success: num=1
retry
デコレータを使って対象の関数をラップするだけでリトライ処理を実装できます。
例外が発生した場合にはリトライが行われ、何も例外が発生しなければ値が返ってきます。
指定した回数だけ繰り返す
指定した回数だけリトライする
from tenacity import retry, stop_after_attempt, RetryError
@retry(stop=stop_after_attempt(3))
def error():
print('Error occured!')
raise Exception('Error occured!')
try:
error()
except RetryError:
print('3 times failed')
# =>
# Error occured!
# Error occured!
# Error occured!
# 3 times failed
retry
デコレータに引数を渡すことでリトライ処理の挙動を変更可能です。
stop_after_attempt
では指定した回数だけリトライを行うことができます。
指定した回数に達した後に例外が発生した場合はRetryError
が発生します。
時間を置いてからリトライする
時間を置いてからリトライする
from datetime import datetime
from tenacity import retry, stop_after_attempt, wait_fixed, RetryError
@retry(stop=stop_after_attempt(3), wait=wait_fixed(3))
def error():
print(f'Error occured!: {datetime.now().strftime("%M:%S")}')
raise Exception('Error occured!')
try:
error()
except RetryError:
print('3 times failed')
# =>
# Error occured!: 06:38
# Error occured!: 06:41
# Error occured!: 06:44
# 3 times failed
ここではwait_fixed
を使って、次のリトライまで3秒待っています。
例えば、APIのレート制限に引っかかって例外が発生した場合などに便利です。
特定の例外が発生した場合にリトライをする
特定の例外が発生した場合にリトライをする
from datetime import datetime
from tenacity import retry, stop_after_attempt, retry_if_exception_type, RetryError
@retry(stop=stop_after_attempt(3), retry=retry_if_exception_type(ValueError))
def error(num):
if num == 0:
print('0 is invalid')
raise ValueError()
raise Exception(f'Error occured!: num={num}')
try:
error(0)
except RetryError:
print('Value Error occured')
# =>
# 0 is invalid
# 0 is invalid
# 0 is invalid
# Value Error occured
try:
error(1)
except Exception as e:
print(e)
# =>
# Error occured!: num=1
retry_if_exception_type
でリトライする場合の例外クラスを指定することができます。
ここでは0が引数として渡された場合はValueError
が発生します。
その場合はリトライ処理が行われます。
一方で1を引数として渡した場合はValueError
以外が発生するのでリトライ処理は行われません。
async/awaitでの例外処理
async/awaitでの例外処理
import asyncio
from tenacity import retry, stop_after_attempt, RetryError
@retry(stop=stop_after_attempt(3))
async def error():
await asyncio.sleep(3)
print('Error occured!')
raise Exception('Error occured!')
try:
loop = asyncio.get_event_loop()
loop.run_until_complete(error())
except RetryError:
print('3 times failed')
# =>
# Error occured!
# Error occured!
# Error occured!
# 3 times failed
tenacityはasync関数にも対応しています。
他の関数と同じようにデコレータでラップすることでリトライ処理を追加できます。
リトライ処理にも失敗した場合には通常の関数と同様にRetryError
が発生します。
終わりに
tenacityを使用することでお手軽にリトライ処理を追加できて便利だと思います。 他にも色々な機能があるので気になった方は公式ドキュメントを見てみてください。