GitHubじゃ!Pythonじゃ!

GitHubからPython関係の優良リポジトリを探したかったのじゃー、でも英語は出来ないから日本語で読むのじゃー、英語社会世知辛いのじゃー

jd

tenacity – Python用のライブラリの再試行

投稿日:

Python用のライブラリの再試行 http://tenacity.readthedocs.io

テナシティ

Tenacityは、Pythonで書かれたApache 2.0ライセンスの汎用再試行ライブラリで、何かにリトライ動作を追加する作業を簡素化します。 これは、再試行のフォークから始まります。

最も単純なユースケースは、値が返されるまで例外が発生するたびにフレーク関数を再試行することです。

.. testcode::

    import random
    from tenacity import retry

    @retry
    def do_something_unreliable():
        if random.randint(0, 10) > 1:
            raise IOError("Broken sauce, everything is hosed!!!111one")
        else:
            return "Awesome sauce!"

    print(do_something_unreliable())

.. testoutput::
   :hide:

   Awesome sauce!

特徴

  • 汎用デコレータAPI
  • 停止条件を指定する(試行回数で制限する)
  • 待機条件を指定する(つまり、試行間の指数バックオフスリープ)
  • 例外の再試行をカスタマイズする
  • 期待される返された結果について再試行をカスタマイズする
  • コルーチンを再試行する

インストール

テナシティをインストールするには、次のようにします。

$ pip install tenacity

基本的な再試行

.. testsetup:: *

    import logging
    from tenacity import *

    class MyException(Exception):
        pass

上で見たように、デフォルトの動作は、例外が発生したときに待たずに永久に再試行することです。

.. testcode::

    @retry
    def never_give_up_never_surrender():
        print("Retry forever ignoring Exceptions, don't wait between retries")
        raise Exception

停止中

少しだけ永続的にして、あきらめる前の試行回数などのいくつかの境界を設定しましょう。

.. testcode::

    @retry(stop=stop_after_attempt(7))
    def stop_after_7_attempts():
        print("Stopping after 7 attempts")
        raise Exception

私たちは一日中はしていないので、何をやり直すべきかの境界を設定しましょう。

.. testcode::

    @retry(stop=stop_after_delay(10))
    def stop_after_10_s():
        print("Stopping after 10 seconds")
        raise Exception

複数の停止条件を結合するには、| オペレーター:

.. testcode::

    @retry(stop=(stop_after_delay(10) | stop_after_attempt(5)))
    def stop_after_10_s_or_5_retries():
        print("Stopping after 10 seconds or 5 retries")
        raise Exception

再試行前に待機中

ほとんどの場合、できるだけ早くポーリングするのが好きではないので、再試行と再試行の間に2秒待ってみましょう。

.. testcode::

    @retry(wait=wait_fixed(2))
    def wait_2_s():
        print("Wait 2 second between retries")
        raise Exception

いくつかのものは、ランダムに注入されたビットで最高のパフォーマンスを発揮します。

.. testcode::

    @retry(wait=wait_random(min=1, max=2))
    def wait_random_1_to_2_s():
        print("Randomly wait 1 to 2 seconds between retries")
        raise Exception

さらに、分散サービスや他のリモートエンドポイントを再試行する際に、指数関数的なバックオフを起こすのは難しいです。

.. testcode::

    @retry(wait=wait_exponential(multiplier=1, max=10))
    def wait_exponential_1():
        print("Wait 2^x * 1 second between each retry, up to 10 seconds, then 10 seconds afterwards")
        raise Exception


また、分散サービスや他のリモートエンドポイントを再試行するときに、一定の待機とジッタを組み合わせて暴動を避けることも困難です。

.. testcode::

    @retry(wait=wait_fixed(3) + wait_random(0, 2))
    def wait_fixed_jitter():
        print("Wait at least 3 seconds, and add up to 2 seconds of random delay")
        raise Exception

複数のプロセスが共有リソースに対して競合する場合、指数関数的に増加するジッタは衝突を最小限に抑えるのに役立ちます。

.. testcode::

    @retry(wait=wait_random_exponential(multiplier=1, max=60))
    def wait_exponential_jitter():
        print("Randomly wait up to 2^x * 1 seconds between each retry until the range reaches 60 seconds, then randomly up to 60 seconds afterwards")
        raise Exception


バックオフの連鎖を構築する必要があることもあります。

.. testcode::

    @retry(wait=wait_chain(*[wait_fixed(3) for i in range(3)] +
                           [wait_fixed(7) for i in range(2)] +
                           [wait_fixed(9)]))
    def wait_fixed_chained():
        print("Wait 3s for 3 attempts, 7s for the next 2 attempts and 9s for all attempts thereafter")
        raise Exception

再試行するかどうか

ここでの場合と同様に、特定の例外または一般的な例外を発生させる再試行を処理するためのオプションがいくつかあります。

.. testcode::

    @retry(retry=retry_if_exception_type(IOError))
    def might_io_error():
        print("Retry forever with no wait if an IOError occurs, raise any other errors")
        raise Exception

また、関数の結果を使用して、再試行の動作を変更することもできます。

.. testcode::

    def is_none_p(value):
        """Return True if value is None"""
        return value is None

    @retry(retry=retry_if_result(is_none_p))
    def might_return_none():
        print("Retry with no wait if return value is None")

いくつかの条件を組み合わせることもできます:

.. testcode::

    def is_none_p(value):
        """Return True if value is None"""
        return value is None

    @retry(retry=(retry_if_result(is_none_p) | retry_if_exception_type()))
    def might_return_none():
        print("Retry forever ignoring Exceptions with no wait if return value is None")

停止、待機などの任意の組み合わせもサポートされており、混在して一致させることができます。

TryAgain例外を発生させることでいつでも明示的に再試行することができます:

.. testcode::

   @retry
   def do_something():
       result = something_else()
       if result == 23:
          raise TryAgain

エラー処理

「タイムアウト」を再試行する呼び出し可能呼び出しは、デフォルトでRetryErrorを発生させますが、必要に応じて最後の試行の例外を再試行できます。

.. testcode::

    @retry(reraise=True, stop=stop_after_attempt(3))
    def raise_my_exception():
        raise MyException("Fail")

    try:
        raise_my_exception()
    except MyException:
        # timed out retrying
        pass

リトライ前後のログ記録

beforeコールバック関数を使って関数を呼び出す前にアクションを実行することは可能です:

.. testcode::

    logger = logging.getLogger(__name__)

    @retry(stop=stop_after_attempt(3), before=before_log(logger, logging.DEBUG))
    def raise_my_exception():
        raise MyException("Fail")

同じ考え方で、失敗した呼び出しの後に実行することができます。

.. testcode::

    logger = logging.getLogger(__name__)

    @retry(stop=stop_after_attempt(3), after=after_log(logger, logging.DEBUG))
    def raise_my_exception():
        raise MyException("Fail")

再試行される障害だけをログに記録することもできます。 通常、再試行は待機間隔の後に行われるため、キーワード引数はbefore_sleepと呼ばれbefore_sleep

.. testcode::

    logger = logging.getLogger(__name__)
    @retry(stop=stop_after_attempt(3),
           before_sleep=before_sleep_log(logger, logging.DEBUG))
    def raise_my_exception():
        raise MyException("Fail")

また、カスタムbefore_sleep関数を定義することもできます。 現在の再試行の呼び出しに関するすべての情報を含むcall_stateという1つのパラメータが必要です。

.. testcode::

    logger = logging.getLogger(__name__)

    def my_before_sleep(call_state):
        if call_state.attempt_number < 1:
            loglevel = logging.INFO
        else:
            loglevel = logging.WARNING
        logger.log(
            loglevel, 'Retrying %s: attempt %s ended with: %s',
            call_state.fn, call_state.attempt_number, call_state.outcome)

    @retry(stop=stop_after_attempt(3), before_sleep=my_before_sleep)
    def raise_my_exception():
        raise MyException("Fail")

    try:
        raise_my_exception()
    except RetryError:
        pass

call_state引数は、RetryCallStateクラスのオブジェクトです。

.. autoclass:: tenacity.RetryCallState

   Constant attributes:

   .. autoinstanceattribute:: start_time(float)
      :annotation:

   .. autoinstanceattribute:: retry_object(BaseRetrying)
      :annotation:

   .. autoinstanceattribute:: fn(callable)
      :annotation:

   .. autoinstanceattribute:: args(tuple)
      :annotation:

   .. autoinstanceattribute:: kwargs(dict)
      :annotation:

   Variable attributes:

   .. autoinstanceattribute:: attempt_number(int)
      :annotation:

   .. autoinstanceattribute:: outcome(tenacity.Future or None)
      :annotation:

   .. autoinstanceattribute:: outcome_timestamp(float or None)
      :annotation:

   .. autoinstanceattribute:: idle_for(float)
      :annotation:

   .. autoinstanceattribute:: next_action(tenacity.RetryAction or None)
      :annotation:



以前のカスタムbefore_sleep関数は、3つのパラメータを受け入れることもできました。 このアプローチは推奨されていませんが、下位互換性のために保持されています。

  • retry_object :再試行を実行するretry_objectオブジェクト。
  • sleep :次の試行までの待機時間(秒)
  • last_result :現在の試行の結果。
.. testcode::

    logger = logging.getLogger(__name__)

    def my_before_sleep(retry_object, sleep, last_result):
        logger.warning(
            'Retrying %s: last_result=%s, retrying in %s seconds...',
            retry_object.fn, last_result, sleep)

    @retry(stop=stop_after_attempt(3), before_sleep=my_before_sleep)
    def raise_my_exception():
        raise MyException("Fail")

    try:
        raise_my_exception()
    except RetryError:
        pass

同様に、すべての再試行が失敗した後、例外を発生させることなくカスタムコールバック関数を呼び出すことができます(または、実際に何かを再実行することも、何かを行うこともできます)

.. testcode::

    def return_last_value(last_attempt):
        """return the result of the last call attempt"""
        return last_attempt.result()

    def is_false(value):
        """Return True if value is False"""
        return value is False

    # will return False after trying 3 times to get a different result
    @retry(stop=stop_after_attempt(3),
           retry_error_callback=return_last_value,
           retry=retry_if_result(is_false))
    def eventually_return_false():
        return False

ファンクションおよび統計情報属性に添付されているretry属性を使用すると、ファンクションに対して実行された再試行に関する統計にアクセスできます。

.. testcode::

    @retry(stop=stop_after_attempt(3))
    def raise_my_exception():
        raise MyException("Fail")

    try:
        raise_my_exception()
    except Exception:
        pass

    print(raise_my_exception.retry.statistics)

.. testoutput::
   :hide:

   ...

実行時の引数の変更

ラップされた関数にアタッチされたretry_with関数を使用して呼び出すときに、必要に応じて再試行デコレータの引数を変更することができます。

.. testcode::

    @retry(stop=stop_after_attempt(3))
    def raise_my_exception():
        raise MyException("Fail")

    try:
        raise_my_exception.retry_with(stop=stop_after_attempt(4))()
    except Exception:
        pass

    print(raise_my_exception.retry.statistics)

.. testoutput::
   :hide:

   ...

再試行パラメータを設定するために変数を使用する場合は、再試行デコレータを使用する必要はありません。代わりに再試行を直接使用することもできます。

.. testcode::

    def never_good_enough(arg1):
        raise Exception('Invalid argument: {}'.format(arg1))

    def try_never_good_enough(max_attempts=3):
        retryer = Retrying(stop=stop_after_attempt(max_attempts), reraise=True)
        retryer(never_good_enough, 'I really do try')

非同期と再試行

最後に、 retryはasyncioとTornado(> = 4.5)コルーチンでも機能します。 スリープは非同期でも行われます。

@retry
async def my_async_function(loop):
    await loop.getaddrinfo('8.8.8.8', 53)
@retry
@tornado.gen.coroutine
def my_async_function(http_client, url):
    yield http_client.fetch(url)

また、正しいスリープ機能を渡すことで、古典やトリオなどの別のイベントループを使用することもできます。

@retry(sleep=trio.sleep)
async def my_async_function(loop):
    await asks.get('https://example.org')

寄稿

  1. 未解決の問題がないかチェックするか、新しく問題を開いて、機能のアイデアやバグに関する議論を開始してください。
  2. GitHub のリポジトリをフォークて、 マスターブランチ(またはブランチからブランチ)への変更を開始します。
  3. バグが修正されたか、または機能が期待どおりに機能することを示すテストを作成します。
  4. ドキュメントをより良くする(またはより詳細にする、読みやすくするなど…)







-jd
-, , ,

執筆者:

jd

tenacity – Python用のライブラリの再試行

投稿日:

(さらに…)







-jd
-, , ,

執筆者: