tenacity

المثابرة

يرجى الرجوع إلى وثائق المثابرة للحصول على تجربة أفضل.

Tenacity هي مكتبة إعادة المحاولة ذات الأغراض العامة المرخصة من Apache 2.0، ومكتوبة بلغة Python، لتبسيط مهمة إضافة سلوك إعادة المحاولة إلى أي شيء تقريبًا. إنه ينشأ من شوكة إعادة المحاولة التي لم يعد يتم صيانتها للأسف. Tenacity غير متوافق مع واجهة برمجة التطبيقات (api) مع إعادة المحاولة ولكنه يضيف وظائف جديدة مهمة ويصلح عددًا من الأخطاء القديمة.

إن أبسط حالة استخدام هي إعادة محاولة دالة غير مستقرة عند حدوث استثناء حتى يتم إرجاع القيمة.

 .. رمز الاختبار::

    استيراد عشوائي
    من مثابرة استيراد إعادة المحاولة

    @إعادة المحاولة
    تعريف do_something_unreliable():
        إذا كان Randint(0, 10) > 1:
            رفع IOError("الصلصة المكسورة، كل شيء مسدود!!!111one")
        آخر:
            العودة "صلصة رائعة!"

    طباعة (do_something_unreliable())

.. إخراج الاختبار::
   :يخفي:

   صلصة رهيبة!

.. توكتري::
    :مختفي:
    :أقصى عمق: 2

    سجل التغيير
    واجهة برمجة التطبيقات

سمات

• واجهة برمجة تطبيقات الديكور العامة
• تحديد شرط التوقف (أي الحد حسب عدد المحاولات)
• تحديد حالة الانتظار (أي التراجع الأسي بين المحاولات)
• تخصيص إعادة المحاولة على الاستثناءات
• تخصيص إعادة المحاولة على النتيجة المتوقعة التي تم إرجاعها
• أعد المحاولة على coroutines
• أعد محاولة كتلة التعليمات البرمجية مع مدير السياق

تثبيت

لتثبيت المثابرة ، ببساطة:

$ pip install tenacity

أمثلة

إعادة المحاولة الأساسية

 .. إعداد الاختبار::

    تسجيل الاستيراد
    #
    # لاحظ أن الاستيراد التالي يُستخدم لتسهيل العرض التوضيحي فقط.
    # يجب أن يقوم كود الإنتاج دائمًا باستيراد الأسماء التي يحتاجها بشكل صريح.
    #
    من استيراد المثابرة *

    فئة MyException (الاستثناء):
        يمر

كما رأيت أعلاه، السلوك الافتراضي هو إعادة المحاولة للأبد دون انتظار ظهور استثناء.

 .. رمز الاختبار::

    @إعادة المحاولة
    بالتأكيد Never_gonna_give_you_up():
        طباعة ("أعد المحاولة مع تجاهل الاستثناءات إلى الأبد، لا تنتظر بين المحاولات")
        رفع الاستثناء

وقف

دعونا نكون أقل إصرارًا ونضع بعض الحدود، مثل عدد المحاولات قبل الاستسلام.

 .. رمز الاختبار::

    @retry(stop=stop_after_attempt(7))
    def stop_after_7_attempts():
        طباعة ("التوقف بعد 7 محاولات")
        رفع الاستثناء

ليس لدينا يوم كامل، لذا دعونا نضع حدًا للمدة التي يجب أن نعيد فيها تجربة الأشياء.

 .. رمز الاختبار::

    @إعادة المحاولة(stop=stop_after_delay(10))
    def stop_after_10_s():
        طباعة ("التوقف بعد 10 ثوانٍ")
        رفع الاستثناء

إذا كان لديك موعد نهائي ضيق، وتجاوز وقت التأخير ليس أمرًا مقبولًا، فيمكنك التخلي عن إعادة المحاولة بمحاولة واحدة قبل أن تتجاوز التأخير.

 .. رمز الاختبار::

    @إعادة المحاولة(stop=stop_before_delay(10))
    def stop_before_10_s():
        طباعة ("إيقاف محاولة واحدة قبل 10 ثوانٍ")
        رفع الاستثناء

يمكنك الجمع بين عدة شروط للتوقف باستخدام | المشغل:

 .. رمز الاختبار::

    @retry(stop=(stop_after_delay(10) | stop_after_attempt(5)))
    def stop_after_10_s_or_5_retries():
        طباعة ("التوقف بعد 10 ثوانٍ أو 5 محاولات")
        رفع الاستثناء

الانتظار قبل إعادة المحاولة

لا ترغب معظم الأشياء في أن يتم استقصاؤها بأسرع ما يمكن، لذا دعنا ننتظر ثانيتين فقط بين مرات إعادة المحاولة.

 .. رمز الاختبار::

    @إعادة المحاولة(الانتظار=wait_fixed(2))
    بالتأكيد wait_2_s ():
        طباعة ("انتظر ثانيتين بين المحاولات")
        رفع الاستثناء

تعمل بعض الأشياء بشكل أفضل مع القليل من العشوائية.

 .. رمز الاختبار::

    @retry(wait=wait_random(min=1, max=2))
    def wait_random_1_to_2_s():
        طباعة ("انتظر عشوائيًا من ثانية إلى ثانيتين بين عمليات إعادة المحاولة")
        رفع الاستثناء

ثم مرة أخرى، من الصعب التغلب على التراجع الأسي عند إعادة محاولة الخدمات الموزعة ونقاط النهاية البعيدة الأخرى.

 .. رمز الاختبار::

    @retry(wait=wait_exponential(multiplier=1, min=4, max=10))
    بالتأكيد wait_exponential_1():
        print("انتظر 2^x * 1 ثانية بين كل إعادة محاولة بدءًا من 4 ثوانٍ، ثم ما يصل إلى 10 ثوانٍ، ثم 10 ثوانٍ بعد ذلك")
        رفع الاستثناء

ومرة أخرى، من الصعب أيضًا التغلب على الجمع بين فترات الانتظار الثابتة والارتعاش (للمساعدة في تجنب القطعان الهادرة) عند إعادة محاولة الخدمات الموزعة ونقاط النهاية البعيدة الأخرى.

 .. رمز الاختبار::

    @retry(wait=wait_fixed(3) + wait_random(0, 2))
    بالتأكيد wait_fixed_jitter():
        طباعة ("انتظر 3 ثوانٍ على الأقل، وأضف ما يصل إلى ثانيتين من التأخير العشوائي")
        رفع الاستثناء

عندما تتنافس عمليات متعددة على مورد مشترك، فإن زيادة الارتعاش بشكل كبير تساعد على تقليل الاصطدامات.

 .. رمز الاختبار::

    @retry(wait=wait_random_exponential(multiplier=1, max=60))
    def wait_exponential_jitter():
        print("انتظر بشكل عشوائي ما يصل إلى 2^x * 1 ثانية بين كل إعادة محاولة حتى يصل النطاق إلى 60 ثانية، ثم بشكل عشوائي حتى 60 ثانية بعد ذلك")
        رفع الاستثناء

في بعض الأحيان يكون من الضروري بناء سلسلة من التراجعات.

 .. رمز الاختبار::

    @retry(wait=wait_chain(*[wait_fixed(3) لـ i في النطاق(3)] +
                           [wait_fixed(7) for i in range(2)] +
                           [انتظار_ثابت(9)]))
    بالتأكيد wait_fixed_chained ():
        طباعة("انتظر 3 ثواني لمدة 3 محاولات، و7 ثواني للمحاولتين التاليتين، و9 ثواني لجميع المحاولات بعد ذلك")
        رفع الاستثناء

سواء لإعادة المحاولة

لدينا بعض الخيارات للتعامل مع عمليات إعادة المحاولة التي تثير استثناءات محددة أو عامة، كما هو الحال في هذه الحالة.

 .. رمز الاختبار::

    خطأ العميل (الاستثناء):
        """أحد أنواع أخطاء العميل."""

    @retry(retry=retry_if_exception_type(IOError))
    بالتأكيد may_io_error():
        print("أعد المحاولة للأبد دون الانتظار في حالة حدوث خطأ IOE، قم برفع أي أخطاء أخرى")
        رفع الاستثناء

    @retry(retry=retry_if_not_exception_type(ClientError))
    بالتأكيد may_client_error():
        print("أعد المحاولة للأبد دون الانتظار في حالة حدوث أي خطأ آخر غير ClientError. ارفع ClientError على الفور.")
        رفع الاستثناء

يمكننا أيضًا استخدام نتيجة الوظيفة لتغيير سلوك إعادة المحاولة.

 .. رمز الاختبار::

    بالتأكيد is_none_p(القيمة):
        """إرجاع صحيح إذا كانت القيمة بلا"""
        قيمة الإرجاع هي لا شيء

    @retry(retry=retry_if_result(is_none_p))
    بالتأكيد may_return_none():
        طباعة ("أعد المحاولة بدون انتظار إذا كانت قيمة الإرجاع لا شيء")

أنظر أيضا هذه الطرق:

 .. رمز الاختبار::

    retry_if_exception
    retry_if_exception_type
    retry_if_not_exception_type
    retry_unless_exception_type
    retry_if_result
    retry_if_not_result
    retry_if_exception_message
    retry_if_not_exception_message
    إعادة المحاولة_any
    إعادة المحاولة_all

يمكننا أيضًا الجمع بين عدة شروط:

 .. رمز الاختبار::

    بالتأكيد is_none_p(القيمة):
        """إرجاع صحيح إذا كانت القيمة بلا"""
        قيمة الإرجاع هي لا شيء

    @retry(retry=(retry_if_result(is_none_p) | retry_if_exception_type()))
    بالتأكيد may_return_none():
        طباعة ("أعد محاولة تجاهل الاستثناءات إلى الأبد دون الانتظار إذا كانت قيمة الإرجاع لا شيء")

يتم أيضًا دعم أي مجموعة من التوقف والانتظار وما إلى ذلك لتمنحك حرية المزج والمطابقة.

من الممكن أيضًا إعادة المحاولة بشكل صريح في أي وقت عن طريق رفع استثناء TryAgain:

 .. رمز الاختبار::

   @إعادة المحاولة
   تعريف do_something():
       النتيجة = شيء_آخر ()
       إذا كانت النتيجة == 23:
          رفع حاول مرة أخرى

معالجة الأخطاء

عادةً عندما تفشل وظيفتك في المرة الأخيرة (ولن تتم إعادة المحاولة مرة أخرى بناءً على إعداداتك)، يتم ظهور خطأ إعادة المحاولة. سيتم عرض الاستثناء الذي واجهته التعليمات البرمجية الخاصة بك في مكان ما في منتصف تتبع المكدس.

إذا كنت تفضل رؤية الاستثناء الذي واجهته التعليمات البرمجية الخاصة بك في نهاية تتبع المكدس (حيث يكون أكثر وضوحًا)، فيمكنك تعيين reraise=True.

 .. رمز الاختبار::

    @retry(reraise=True, stop=stop_after_attempt(3))
    Def rise_my_exception():
        رفع MyException("فشل")

    يحاول:
        rise_my_exception()
    باستثناء MyException:
        # انتهت مهلة إعادة المحاولة
        يمر

قبل وبعد إعادة المحاولة والتسجيل

من الممكن تنفيذ إجراء قبل أي محاولة لاستدعاء الوظيفة باستخدام وظيفة رد الاتصال before :

 .. رمز الاختبار::

    تسجيل الاستيراد
    نظام الاستيراد

    logging.basicConfig(stream=sys.stderr,level=logging.DEBUG)

    المسجل = logger.getLogger(__name__)

    @retry(stop=stop_after_attempt(3), before=before_log(logger,logging.DEBUG))
    Def rise_my_exception():
        رفع MyException("فشل")

وبنفس الروح، من الممكن التنفيذ بعد فشل المكالمة:

 .. رمز الاختبار::

    تسجيل الاستيراد
    نظام الاستيراد

    logging.basicConfig(stream=sys.stderr,level=logging.DEBUG)

    المسجل = logging.getLogger(__name__)

    @retry(stop=stop_after_attempt(3), after=after_log(logger,logging.DEBUG))
    Def rise_my_exception():
        رفع MyException("فشل")

من الممكن أيضًا تسجيل حالات الفشل التي ستتم إعادة محاولتها فقط. عادةً ما تتم عمليات إعادة المحاولة بعد فترة انتظار، لذلك يتم استدعاء وسيطة الكلمة الأساسية before_sleep :

 .. رمز الاختبار::

    تسجيل الاستيراد
    نظام الاستيراد

    logging.basicConfig(stream=sys.stderr,level=logging.DEBUG)

    المسجل = logging.getLogger(__name__)

    @إعادة المحاولة(إيقاف=stop_after_attempt(3),
           before_sleep=before_sleep_log(المسجل، logging.DEBUG))
    Def rise_my_exception():
        رفع MyException("فشل")

إحصائيات

يمكنك الوصول إلى الإحصائيات حول إعادة المحاولة التي تم إجراؤها عبر إحدى الوظائف باستخدام سمة الإحصائيات المرفقة بالوظيفة:

 .. رمز الاختبار::

    @retry(stop=stop_after_attempt(3))
    Def rise_my_exception():
        رفع MyException("فشل")

    يحاول:
        rise_my_exception()
    باستثناء الاستثناء:
        يمر

    طباعة (raise_my_exception.statistics)

.. إخراج الاختبار::
   :يخفي:

   ...

عمليات الاسترجاعات المخصصة

يمكنك أيضًا تحديد عمليات الاسترجاعات الخاصة بك. يجب أن يقبل رد الاتصال معلمة واحدة تسمى retry_state والتي تحتوي على كافة المعلومات حول استدعاء إعادة المحاولة الحالي.

على سبيل المثال، يمكنك استدعاء وظيفة رد اتصال مخصصة بعد فشل جميع عمليات إعادة المحاولة، دون رفع استثناء (أو يمكنك إعادة الرفع أو القيام بأي شيء بالفعل)

 .. رمز الاختبار::

    def return_last_value(retry_state):
        """إرجاع نتيجة آخر محاولة اتصال"""
        إرجاع retry_state.outcome.result()

    تعريف is_false(القيمة):
        """إرجاع صحيح إذا كانت القيمة خاطئة"""
        قيمة الإرجاع خاطئة

    # سيعود خطأ بعد المحاولة 3 مرات للحصول على نتيجة مختلفة
    @إعادة المحاولة(إيقاف=stop_after_attempt(3),
           retry_error_callback=return_last_value،
           إعادة المحاولة=retry_if_result(is_false))
    بالتأكيد_return_false():
        العودة كاذبة

إعادة محاولة CallState

تعد وسيطة retry_state كائنًا من الفئة :class:`~tenacity.RetryCallState` .

عمليات الاسترجاعات المخصصة الأخرى

من الممكن أيضًا تحديد عمليات الاسترجاعات المخصصة لوسائط الكلمات الرئيسية الأخرى.

 .. الوظيفة:: my_stop(retry_state)

   :param RetryCallState retry_state: معلومات حول استدعاء إعادة المحاولة الحالي
   :return: ما إذا كان يجب أن تتوقف إعادة المحاولة أم لا
   :rtype: منطقي

.. الوظيفة:: my_wait(retry_state)

   :param RetryCallState retry_state: معلومات حول استدعاء إعادة المحاولة الحالي
   :return: عدد الثواني التي يجب الانتظار قبل إعادة المحاولة التالية
   :rtype: تعويم

.. الوظيفة:: my_retry(retry_state)

   :param RetryCallState retry_state: معلومات حول استدعاء إعادة المحاولة الحالي
   :return: ما إذا كان ينبغي الاستمرار في إعادة المحاولة أم لا
   :rtype: منطقي

.. الوظيفة:: my_before(retry_state)

   :param RetryCallState retry_state: معلومات حول استدعاء إعادة المحاولة الحالي

.. الوظيفة:: my_after(retry_state)

   :param RetryCallState retry_state: معلومات حول استدعاء إعادة المحاولة الحالي

.. الوظيفة:: my_before_sleep(retry_state)

   :param RetryCallState retry_state: معلومات حول استدعاء إعادة المحاولة الحالي

فيما يلي مثال لوظيفة before_sleep المخصصة:

 .. رمز الاختبار::

    تسجيل الاستيراد

    logging.basicConfig(stream=sys.stderr,level=logging.DEBUG)

    المسجل = logging.getLogger(__name__)

    حدد my_before_sleep(retry_state):
        إذا كان retry_state.attempt_number < 1:
            مستوى السجل = تسجيل الدخول.INFO
        آخر:
            مستوى السجل = تسجيل.تحذير
        المسجل.سجل(
            مستوى السجل، 'إعادة محاولة %s: انتهت المحاولة %s بـ: %s'،
            retry_state.fn، retry_state.attempt_number، retry_state.outcome)

    @retry(stop=stop_after_attempt(3), before_sleep=my_before_sleep)
    Def rise_my_exception():
        رفع MyException("فشل")

    يحاول:
        rise_my_exception()
    باستثناء خطأ إعادة المحاولة:
        يمر

تغيير الوسائط في وقت التشغيل

يمكنك تغيير وسيطات مصمم إعادة المحاولة حسب الحاجة عند استدعائه باستخدام الدالة retry_with المرفقة بالوظيفة الملتفة:

 .. رمز الاختبار::

    @retry(stop=stop_after_attempt(3))
    Def rise_my_exception():
        رفع MyException("فشل")

    يحاول:
        rise_my_exception.retry_with(stop=stop_after_attempt(4))()
    باستثناء الاستثناء:
        يمر

    طباعة (raise_my_exception.statistics)

.. إخراج الاختبار::
   :يخفي:

   ...

إذا كنت تريد استخدام المتغيرات لإعداد معاملات إعادة المحاولة، فلن تحتاج إلى استخدام مصمم إعادة المحاولة - يمكنك بدلاً من ذلك استخدام إعادة المحاولة مباشرةً:

 .. رمز الاختبار::

    بالتأكيد Never_good_enough(arg1):
        رفع الاستثناء ("وسيطة غير صالحة: {}". التنسيق (arg1))

    بالتأكيد حاول_never_good_enough(max_attempts=3):
        إعادة المحاولة = إعادة المحاولة (التوقف=stop_after_attempt(max_attempts)، إعادة المحاولة=True)
        إعادة المحاولة (never_good_enough، "أنا أحاول حقًا")

قد ترغب أيضًا في تغيير سلوك وظيفة مزخرفة مؤقتًا، كما هو الحال في الاختبارات لتجنب أوقات الانتظار غير الضرورية. يمكنك تعديل/تصحيح سمة إعادة المحاولة المرفقة بالوظيفة. ضع في اعتبارك أن هذه سمة للكتابة فقط، ويجب قراءة الإحصائيات من سمة إحصائيات الوظيفة.

 .. رمز الاختبار::

    @retry(stop=stop_after_attempt(3), wait=wait_fixed(3))
    Def rise_my_exception():
        رفع MyException("فشل")

    من نموذج الاستيراد Unittest

    باستخدام mock.patch.object(raise_my_exception.retry, "wait", wait_fixed(0)):
        يحاول:
            rise_my_exception()
        باستثناء الاستثناء:
            يمر

    طباعة (raise_my_exception.statistics)

.. إخراج الاختبار::
   :يخفي:

   ...

إعادة محاولة كتلة التعليمات البرمجية

تتيح لك Tenacity إعادة محاولة كتلة التعليمات البرمجية دون الحاجة إلى تغليفها في وظيفة معزولة. وهذا يجعل من السهل عزل الكتلة الفاشلة أثناء مشاركة السياق. الحيلة هي الجمع بين حلقة for ومدير السياق.

 .. رمز الاختبار::

   من إعادة المحاولة لاستيراد المثابرة، وRetryError، وstop_after_attempt

   يحاول:
       لمحاولة إعادة المحاولة (stop=stop_after_attempt(3)):
           مع المحاولة:
               رفع الاستثناء ("الرمز الخاص بي فشل!")
   باستثناء خطأ إعادة المحاولة:
       يمر

يمكنك تكوين كل تفاصيل سياسة إعادة المحاولة عن طريق تكوين كائن إعادة المحاولة.

باستخدام التعليمات البرمجية غير المتزامنة، يمكنك استخدام AsyncRetrying.

 .. رمز الاختبار::

   من المثابرة استيراد AsyncRetrying، RetryError، stop_after_attempt

   دالة التعريف غير المتزامنة ():
      يحاول:
          غير متزامن للمحاولة في AsyncRetrying(stop=stop_after_attempt(3)):
              مع المحاولة:
                  رفع الاستثناء ("الرمز الخاص بي فشل!")
      باستثناء خطأ إعادة المحاولة:
          يمر

في كلتا الحالتين، قد ترغب في تعيين النتيجة للمحاولة بحيث تكون متاحة في إستراتيجيات إعادة المحاولة مثل retry_if_result . يمكن القيام بذلك من خلال الوصول إلى خاصية retry_state :

 .. رمز الاختبار::

    من المثابرة استيراد AsyncRetrying، retry_if_result

    دالة التعريف غير المتزامنة ():
       غير متزامن لمحاولة AsyncRetrying(retry=retry_if_result(lambda x: x < 3)):
           مع المحاولة:
               النتيجة = 1 # بعض الحسابات المعقدة واستدعاء الوظائف وما إلى ذلك.
           إذا لم يكن الأمر كذلك، حاول إعادة المحاولة_state.outcome.failed:
               محاولة.retry_state.set_result(نتيجة)
       نتيجة العودة

غير متزامن وأعد المحاولة

وأخيرًا، تعمل retry أيضًا على كوروتينات asyncio وTrio وTornado (>= 4.5). تتم عمليات النوم بشكل غير متزامن أيضًا.

 @ retry
async def my_asyncio_function ( loop ):
    await loop . getaddrinfo ( '8.8.8.8' , 53 )

 @ retry
async def my_async_trio_function ():
    await trio . socket . getaddrinfo ( '8.8.8.8' , 53 )

 @ retry
@ tornado . gen . coroutine
def my_async_tornado_function ( http_client , url ):
    yield http_client . fetch ( url )

يمكنك أيضًا استخدام حلقات أحداث بديلة مثل التحف عن طريق تمرير وظيفة السكون الصحيحة:

 @ retry ( sleep = curio . sleep )
async def my_async_curio_function ():
    await asks . get ( 'https://example.org' )

يساهم

1. تحقق من وجود مشكلات مفتوحة أو افتح مشكلة جديدة لبدء مناقشة حول فكرة ميزة أو خطأ.
2. قم بتفرع المستودع على GitHub لبدء إجراء التغييرات على الفرع الرئيسي (أو الفرع خارجه).
3. اكتب اختبارًا يوضح أنه تم إصلاح الخلل أو أن الميزة تعمل كما هو متوقع.
4. أضف سجل التغيير
5. اجعل المستندات أفضل (أو أكثر تفصيلاً، أو أسهل في القراءة، أو ...)

سجلات التغيير

يستخدم رينو لإدارة سجلات التغيير. ألق نظرة على مستندات الاستخدام الخاصة بهم.

سيقوم إنشاء المستندات بتجميع سجلات التغيير تلقائيًا. تحتاج فقط إلى إضافتها.

 # Opens a template file in an editor
tox -e reno -- new some-slug-for-my-change --edit