Pythonを使用してTenacityを使用してリトライを制御する：エンドツーエンドのチュートリアル

PythonでTenacityを使ってリトライを制御する方法をエンドツーエンドで学ぶためのチュートリアル

PYTHON TOOLBOX（パイソンツールボックス）

頑強な再試行メカニズムとエラーハンドリング技術を使ってPythonプロジェクトを強化する

この記事では、Tenacityの基本的な使用方法とカスタマイズ機能について説明します。

この便利なPythonライブラリは、再試行メカニズムを提供します。また、具体的な例を通じてTenacityの再試行と例外処理の機能も探求します。

はじめに

海外にある（レイテンシが高い）ウェブサービスや非常に古い（そして安定していない）ウェブサービスを管理していると想像してみてください。どう感じますか？

私の同僚のワンさんはこのような窮状にあると言っていました。彼はかなりイライラしていました：

• Snowflake上でストリーミング半構造化データ分析プラットフォームを構築する方法
• データサイエンスは良いキャリアですか？
• データアナリストからデータサイエンティストへのキャリアチェンジの方法は？

彼は毎日、これらのリモートサービスの呼び出し状況をチェックする必要があり、タイムアウトの問題やその他の異常に頻繁に遭遇します。トラブルシューティングは特に困難です。

さらに、多くのクライアント側のコードは彼の前任者によって書かれており、大規模なリファクタリングを行うことが困難です。したがって、サービスは現状のままで続ける必要があります。

例外が発生した後にこれらのリモート呼び出しを自動的に再接続する方法があれば便利です。ワンさんは涙を浮かべながら私を見ました。

私は彼に心配いりませんと言い、私のツールボックスから新しいツールTenacityを紹介しました。わずか1つのデコレータを追加するだけで、既存のコードに再試行機能を追加することができます。使い方を見てみましょう。

インストールと基本的な使用方法

Tenacityの公式ウェブサイトでは簡単なAPIドキュメントしか提供されていないため、まずライブラリのインストールと基本的な使用方法を見てみましょう。

インストール

pipを使用している場合は、次のコマンドを実行します：

python -m pip install tenacity

Anacondaを使用している場合、Tenacityはデフォルトのチャネルには存在しないため、conda-forgeからインストールする必要があります：

conda install -c conda-forge tenacity

基本的な使用方法

Tenacityをインストールした後、ライブラリの基本的な使用方法を見てみましょう。

@retryデコレータを追加するだけで、コードに再試行機能が追加されます：

@retry()async def coro_func():    pass

一定回数の試行後に再試行を停止する場合は、次のように記述できます：

@retry(stop=stop_after_attempt(5))async def coro_func():    pass

もちろん、接続プールを枯渇させる可能性のある頻繁な再試行を避けるために、各再試行の前に待機時間を追加することをおすすめします。たとえば、各接続の前に2秒待機する場合：

@retry(wait=wait_fixed(2))async def coro_func():    pass

ドキュメントには記載されていませんが、前回よりも1秒長く待機してから各再試行を行うことで、リソースの浪費を最小限に抑えることができます：

@retry(wait=wait_incrementing(start=1, increment=1, max=5))async def coro_func():    pass

最後に、メソッド内で発生した例外によって再試行が引き起こされた場合、例外を再度スローすることが最善です。これにより、メソッドを呼び出す際の柔軟な例外処理が可能になります：

@retry(reraise=True, stop=stop_after_attempt(3))async def coro_func():    pass

高度な機能：カスタムコールバック

一般的な使用例に加えて、メソッドの実行結果に基づいた再試行判定ロジックの追加や、メソッド呼び出し前にメソッドの引数を表示するなど、独自の再試行判定ロジックを追加することができます。

この場合、カスタムコールバックを使用してカスタマイズを行います。

Custom Callbacksを拡張する方法は2つあります：

1つ目は、ドキュメントで推奨されている方法で、拡張メソッドを記述することです。

このメソッドは、実行時にRetryCallStateインスタンスがパラメータとして渡されます。

このパラメータを通じて、ラップされたメソッド、メソッド呼び出しのパラメータ、返された結果、および発生した例外を取得することができます。

例えば、このアプローチを使用して、メソッドの戻り値を判断し、値が偶数の場合には再試行することができます：

from tenacity import *def check_is_even(retry_state: RetryCallState):    if retry_state.outcome.exception():        return True    return retry_state.outcome.result() % 2 == 0

もちろん、この判断を行う前にexceptionがスローされた場合は、直接再試行します。

拡張メソッドに追加のパラメータを渡す必要がある場合は、拡張メソッドの外側にラッパーを追加することができます。

例えば、このラッパーはloggerパラメータを渡します。再試行回数が2回を超えると、ログに再試行の回数、メソッド名、およびメソッドパラメータを出力します：

def my_before_log(logger: Logger):    def my_log(retry_state: RetryCallState):        fn = retry_state.fn        args = retry_state.args        attempt = retry_state.attempt_number        if attempt > 2:            logger.warning(f"Start retry method {fn.__name__} with args: {args}")    return my_log

リアルワールドのネットワークの例

最後に、プロジェクトでを使用する際の強力な機能を実演するために、リモートクライアントプロジェクトを例に挙げます。

このプロジェクトでは、HTTPサービスへのアクセスをシミュレートし、返されたstatus codeに基づいて再試行するかどうかを決定します。

もちろん、長時間の接続待ち時間によるサーバーリソースの浪費を避けるために、各リクエストに2秒のタイムアウトを追加します。タイムアウトが発生した場合は、接続を再試行します。

コードを開始する前に、いくつかの拡張メソッドを実装します。1つのメソッドは、メソッドの再試行回数が2回を超えた場合に警告メッセージをログに出力することです：

もう1つの拡張メソッドは、返されたstatus codeを判断することです。ステータスコードが300より大きい場合は再試行します。もちろん、タイムアウトも再試行をトリガーします。

次に、リモートコールメソッドの実装があります。メソッドを書いたら、Tenacityのリトライデコレータを追加するのを忘れないでください。ここで使用する戦略は、前のリトライよりも1秒長く待機して、最大20回リトライすることです。

もちろん、さきほど実装した2つの拡張メソッドも追加することを忘れないでください：

ミッション完了！とても簡単でしたね？

結論

また、Wangさんの問題を解決するのに役立てることができて嬉しいです。

を使用することで、既存のコードにさまざまなリトライメカニズムを簡単に組み込むことができ、プログラムの堅牢性と自己回復能力を向上させることができます。

このライブラリが問題解決のお手伝いになれば、さらに嬉しいです。コメントを残して議論に参加してください。

コードの実行速度とパフォーマンスを向上させることに加えて、さまざまなツールを使用して作業効率を向上させることもパフォーマンスの向上につながります：

Peng Qian

Pythonツールボックス

4つのストーリーのリストを表示する

紹介リンクを使用してVoAGIに参加してください – Peng Qian

VoAGIメンバーとして、お支払いいただく会費の一部が読んだ作家に届き、すべてのストーリーに完全アクセスできます…

VoAGI.com

We will continue to update VoAGI; if you have any questions or suggestions, please contact us!