tenacity

ความดื้อรั้น

โปรดดู เอกสารประกอบการดื้อรั้น เพื่อประสบการณ์ที่ดีขึ้น

Tenacity เป็นไลบรารีการลองซ้ำสำหรับจุดประสงค์ทั่วไปที่ได้รับใบอนุญาต Apache 2.0 ซึ่งเขียนด้วย Python เพื่อลดความซับซ้อนของการเพิ่มพฤติกรรมการลองซ้ำให้กับทุกสิ่ง มันมาจากทางแยกของการลองใหม่ซึ่งน่าเสียดายที่ไม่ได้รับการดูแลอีกต่อไป Tenacity เข้ากันไม่ได้กับ API กับการลองใหม่ แต่เพิ่มฟังก์ชันใหม่ที่สำคัญและแก้ไขข้อบกพร่องที่มีมายาวนานจำนวนหนึ่ง

กรณีการใช้งานที่ง่ายที่สุดคือการลองใช้ฟังก์ชันที่ไม่สม่ำเสมออีกครั้งทุกครั้งที่มีข้อยกเว้นเกิดขึ้นจนกว่าจะส่งคืนค่า

 .. รหัสทดสอบ::

    นำเข้าแบบสุ่ม
    จากการลองนำเข้าความดื้อรั้นอีกครั้ง

    @ลองอีกครั้ง
    def do_something_unreliable ():
        ถ้าสุ่ม randint (0, 10) > 1:
            Raise IOError("ซอสแตก ทุกอย่างพัง!!!111one")
        อื่น:
            กลับมา "น้ำจิ้มเด็ด!"

    พิมพ์ (do_something_unreliable())

.. ทดสอบผลลัพธ์::
   :ซ่อน:

   ซอสเด็ด!

.. ท็อกทรี::
    :ที่ซ่อนอยู่:
    :ความลึกสูงสุด: 2

    บันทึกการเปลี่ยนแปลง
    เอพีไอ

คุณสมบัติ

• API มัณฑนากรทั่วไป
• ระบุเงื่อนไขการหยุด (เช่น จำกัดตามจำนวนครั้ง)
• ระบุเงื่อนไขการรอ (เช่น การสลีปแบ็คออฟแบบเอกซ์โปเนนเชียลระหว่างความพยายาม)
• ปรับแต่งการลองใหม่อีกครั้งบนข้อยกเว้น
• ปรับแต่งการลองใหม่ตามผลลัพธ์ที่คาดหวัง
• ลองใช้โครูทีนอีกครั้ง
• ลองบล็อกโค้ดอีกครั้งด้วยตัวจัดการบริบท

การติดตั้ง

หากต้องการติดตั้ง tenacity เพียง:

$ pip install tenacity

ตัวอย่าง

ลองใหม่ขั้นพื้นฐาน

 .. ตั้งค่าการทดสอบ::

    การบันทึกการนำเข้า
    -
    # โปรดทราบว่าการนำเข้าต่อไปนี้ใช้เพื่อความสะดวกในการสาธิตเท่านั้น
    # รหัสการผลิตควรนำเข้าชื่อที่ต้องการอย่างชัดเจนเสมอ
    -
    จากการนำเข้าความดื้อรั้น *

    คลาส MyException (ข้อยกเว้น):
        ผ่าน

ดังที่คุณเห็นข้างต้น พฤติกรรมเริ่มต้นคือการลองใหม่ตลอดไปโดยไม่ต้องรอเมื่อมีการหยิบยกข้อยกเว้นขึ้นมา

 .. รหัสทดสอบ::

    @ลองอีกครั้ง
    def never_gonna_give_you_up ():
        print("ลองใหม่โดยไม่สนใจข้อยกเว้น ไม่ต้องรอระหว่างการลองใหม่")
        เพิ่มข้อยกเว้น

กำลังหยุด

อดทนน้อยลงสักหน่อยและกำหนดขอบเขต เช่น จำนวนครั้งที่พยายามก่อนที่จะยอมแพ้

 .. รหัสทดสอบ::

    @ลองอีกครั้ง(หยุด=stop_after_attempt(7))
    def stop_after_7_attempts ():
        print("หยุดหลังจากพยายาม 7 ครั้ง")
        เพิ่มข้อยกเว้น

เราไม่มีเวลาทั้งวัน ดังนั้นเรามากำหนดขอบเขตว่าเราควรลองทำอะไรซ้ำๆ กันนานแค่ไหน

 .. รหัสทดสอบ::

    @ลองอีกครั้ง(หยุด=stop_after_delay(10))
    def stop_after_10_s():
        print("หยุดหลังจาก 10 วินาที")
        เพิ่มข้อยกเว้น

หากคุณมีกำหนดเวลาที่จำกัด และเกินเวลาล่าช้านั้นไม่เป็นไร คุณสามารถละทิ้งความพยายามอีกครั้งหนึ่งก่อนที่จะเกินความล่าช้า

 .. รหัสทดสอบ::

    @ลองอีกครั้ง(หยุด=stop_before_delay(10))
    def stop_before_10_s():
        print("หยุด 1 ครั้งก่อน 10 วินาที")
        เพิ่มข้อยกเว้น

คุณสามารถรวมเงื่อนไขการหยุดหลายรายการได้โดยใช้ | ตัวดำเนินการ:

 .. รหัสทดสอบ::

    @retry(stop=(stop_after_delay(10) | stop_after_attempt(5)))
    def stop_after_10_s_or_5_ลองใหม่ ():
        print("หยุดหลังจาก 10 วินาทีหรือลองใหม่ 5 ครั้ง")
        เพิ่มข้อยกเว้น

รอก่อนที่จะลองอีกครั้ง

สิ่งต่างๆ ส่วนใหญ่ไม่ชอบให้มีการโพลเร็วที่สุดเท่าที่จะเป็นไปได้ ดังนั้นให้รอ 2 วินาทีระหว่างการลองใหม่อีกครั้ง

 .. รหัสทดสอบ::

    @ลองอีกครั้ง(รอ=รอ_แก้ไข(2))
    def wait_2_s():
        print("รอ 2 วินาทีระหว่างการลองใหม่")
        เพิ่มข้อยกเว้น

บางสิ่งจะทำงานได้ดีที่สุดโดยมีการสุ่มเข้าไปเล็กน้อย

 .. รหัสทดสอบ::

    @retry(รอ=รอ_สุ่ม(นาที=1, สูงสุด=2))
    def wait_random_1_to_2_s():
        print("สุ่มรอ 1 ถึง 2 วินาทีระหว่างการลองใหม่")
        เพิ่มข้อยกเว้น

อีกครั้งหนึ่ง เป็นการยากที่จะเอาชนะการถอยกลับแบบเอ็กซ์โปเนนเชียลเมื่อลองใช้บริการแบบกระจายและอุปกรณ์ปลายทางระยะไกลอื่นๆ อีกครั้ง

 .. รหัสทดสอบ::

    @retry(รอ=wait_exponential(ตัวคูณ=1, นาที=4, สูงสุด=10))
    def wait_exponential_1():
        print("รอ 2^x * 1 วินาทีระหว่างการลองแต่ละครั้งโดยเริ่มจาก 4 วินาที จากนั้นสูงสุด 10 วินาที จากนั้น 10 วินาทีหลังจากนั้น")
        เพิ่มข้อยกเว้น

อีกครั้งหนึ่ง ยังยากที่จะเอาชนะการรวมการรอคงที่และความกระวนกระวายใจ (เพื่อช่วยหลีกเลี่ยงฝูงสัตว์ที่ฟ้าร้อง) เมื่อลองบริการแบบกระจายและจุดสิ้นสุดระยะไกลอื่นๆ อีกครั้ง

 .. รหัสทดสอบ::

    @retry(รอ=รอ_แก้ไข(3) + wait_random(0, 2))
    def wait_fixed_jitter ():
        print("รออย่างน้อย 3 วินาทีและเพิ่มความล่าช้าแบบสุ่มสูงสุด 2 วินาที")
        เพิ่มข้อยกเว้น

เมื่อหลายกระบวนการขัดแย้งกันสำหรับทรัพยากรที่ใช้ร่วมกัน ความกระวนกระวายใจที่เพิ่มขึ้นแบบทวีคูณจะช่วยลดความขัดแย้งให้เหลือน้อยที่สุด

 .. รหัสทดสอบ::

    @retry(รอ=wait_random_exponential(ตัวคูณ=1,สูงสุด=60))
    def wait_exponential_jitter():
        print("สุ่มรอสูงสุด 2^x * 1 วินาทีระหว่างการลองแต่ละครั้งจนกระทั่งช่วงถึง 60 วินาที จากนั้นสุ่มสูงสุด 60 วินาทีหลังจากนั้น")
        เพิ่มข้อยกเว้น

บางครั้งจำเป็นต้องสร้างห่วงโซ่ของการถอยกลับ

 .. รหัสทดสอบ::

    @retry(wait=wait_chain(*[wait_fixed(3) for i in range(3)] +
                           [wait_fixed(7) สำหรับฉันอยู่ในช่วง (2)] +
                           [wait_fixed(9)]))
    def wait_fixed_chained ():
        print("รอ 3 วินาทีสำหรับ 3 ครั้ง, 7 วินาทีสำหรับ 2 ครั้งถัดไป และ 9 วินาทีสำหรับความพยายามทั้งหมดหลังจากนั้น")
        เพิ่มข้อยกเว้น

ไม่ว่าจะลองอีกครั้ง

เรามีทางเลือกสองสามทางในการจัดการกับการลองใหม่ซึ่งทำให้เกิดข้อยกเว้นเฉพาะหรือทั่วไป ดังในกรณีต่อไปนี้

 .. รหัสทดสอบ::

    คลาส ClientError (ข้อยกเว้น):
        """ข้อผิดพลาดไคลเอ็นต์บางประเภท"""

    @retry(retry=retry_if_Exception_type(IOError))
    def might_io_error():
        print("ลองใหม่ตลอดไปโดยไม่ต้องรอหากเกิด IOError แจ้งข้อผิดพลาดอื่นๆ")
        เพิ่มข้อยกเว้น

    @retry(retry=retry_if_not_Exception_type(ClientError))
    def might_client_error ():
        print("ลองใหม่ตลอดไปโดยไม่ต้องรอหากมีข้อผิดพลาดอื่นนอกเหนือจาก ClientError เกิดขึ้น ให้เพิ่ม ClientError ทันที")
        เพิ่มข้อยกเว้น

นอกจากนี้เรายังสามารถใช้ผลลัพธ์ของฟังก์ชันเพื่อปรับเปลี่ยนพฤติกรรมการลองใหม่ได้

 .. รหัสทดสอบ::

    def is_none_p (ค่า):
        """คืนค่าเป็นจริงถ้าค่าเป็นไม่มี"""
        ค่าส่งคืนคือไม่มี

    @retry(retry=retry_if_result(is_none_p))
    def might_return_none ():
        print("ลองใหม่โดยไม่ต้องรอถ้าค่าที่ส่งคืนเป็น None")

ดูวิธีการเหล่านี้ด้วย:

 .. รหัสทดสอบ::

    ลองอีกครั้ง_if_ข้อยกเว้น
    ลองใหม่_if_Exception_type
    ลองใหม่อีกครั้ง_if_not_Exception_type
    ลองใหม่อีกครั้ง_unless_Exception_type
    ลองใหม่_if_result
    ลองอีกครั้ง_if_not_result
    ลองใหม่_if_Exception_message
    ลองใหม่อีกครั้ง_if_not_Exception_message
    ลองอีกครั้ง_ใดๆ
    ลองอีกครั้ง_ทั้งหมด

นอกจากนี้เรายังสามารถรวมเงื่อนไขหลายประการเข้าด้วยกัน:

 .. รหัสทดสอบ::

    def is_none_p (ค่า):
        """คืนค่าเป็นจริงถ้าค่าเป็นไม่มี"""
        ค่าส่งคืนคือไม่มี

    @retry(retry=(retry_if_result(is_none_p) | retry_if_Exception_type()))
    def might_return_none ():
        print("ลองใหม่โดยไม่สนใจข้อยกเว้นโดยไม่ต้องรอหากค่าที่ส่งคืนเป็น None")

รองรับการผสมผสานระหว่างการหยุด รอ ฯลฯ เพื่อให้คุณมีอิสระในการมิกซ์แอนด์แมตช์

นอกจากนี้ยังสามารถลองอีกครั้งอย่างชัดเจนเมื่อใดก็ได้โดยเพิ่มข้อยกเว้น TryAgain:

 .. รหัสทดสอบ::

   @ลองอีกครั้ง
   def ทำ_บางอย่าง ():
       ผลลัพธ์ = some_else()
       ถ้าผลลัพธ์ == 23:
          ยกลองอีกครั้ง

การจัดการข้อผิดพลาด

โดยปกติเมื่อฟังก์ชันของคุณล้มเหลวในครั้งสุดท้าย (และจะไม่ลองอีกครั้งตามการตั้งค่าของคุณ) RetryError จะเกิดขึ้น ข้อยกเว้นที่โค้ดของคุณพบจะแสดงอยู่ที่ไหนสักแห่ง ตรงกลาง ของการติดตามสแต็ก

หากคุณต้องการเห็นข้อยกเว้นที่โค้ดของคุณพบที่ ส่วนท้าย ของการติดตามสแต็ก (ซึ่งมองเห็นได้ชัดเจนที่สุด) คุณสามารถตั้งค่า reraise=True

 .. รหัสทดสอบ::

    @retry(reraise=True, stop=stop_after_attempt(3))
    def Raise_my_Exception():
        เพิ่ม MyException("ล้มเหลว")

    พยายาม:
        Raise_my_Exception()
    ยกเว้น MyException:
        # หมดเวลาลองอีกครั้ง
        ผ่าน

ก่อนและหลังลองอีกครั้ง และการบันทึก

คุณสามารถดำเนินการก่อนที่จะพยายามเรียกใช้ฟังก์ชันโดยใช้ฟังก์ชัน before callback:

 .. รหัสทดสอบ::

    การบันทึกการนำเข้า
    ระบบนำเข้า

    logging.basicConfig (สตรีม = sys.stderr, ระดับ = logging. DEBUG)

    คนตัดไม้ = logging.getLogger(__name__)

    @retry(stop=stop_after_attempt(3), before=before_log(คนตัดไม้, logging.DEBUG))
    def Raise_my_Exception():
        เพิ่ม MyException("ล้มเหลว")

ด้วยจิตวิญญาณเดียวกัน คุณสามารถดำเนินการหลังจากการเรียกที่ล้มเหลว:

 .. รหัสทดสอบ::

    การบันทึกการนำเข้า
    ระบบนำเข้า

    logging.basicConfig (สตรีม = sys.stderr, ระดับ = logging. DEBUG)

    คนตัดไม้ = logging.getLogger(__name__)

    @retry(stop=stop_after_attempt(3), after=after_log(logger, logging.DEBUG))
    def Raise_my_Exception():
        เพิ่ม MyException("ล้มเหลว")

นอกจากนี้ยังสามารถบันทึกเฉพาะความล้มเหลวที่จะพยายามอีกครั้งเท่านั้น โดยปกติการลองใหม่จะเกิดขึ้นหลังจากช่วงเวลารอ ดังนั้นอาร์กิวเมนต์ของคำหลักจึงถูกเรียกว่า before_sleep :

 .. รหัสทดสอบ::

    การบันทึกการนำเข้า
    ระบบนำเข้า

    logging.basicConfig (สตรีม = sys.stderr, ระดับ = logging. DEBUG)

    คนตัดไม้ = logging.getLogger(__name__)

    @retry(stop=stop_after_attempt(3),
           before_sleep=before_sleep_log (คนตัดไม้, logging.DEBUG))
    def Raise_my_Exception():
        เพิ่ม MyException("ล้มเหลว")

สถิติ

คุณสามารถเข้าถึงสถิติเกี่ยวกับการลองใหม่บนฟังก์ชันได้โดยใช้แอตทริบิวต์ Statistics ที่แนบมากับฟังก์ชัน:

 .. รหัสทดสอบ::

    @ลองอีกครั้ง(หยุด=stop_after_attempt(3))
    def Raise_my_Exception():
        เพิ่ม MyException("ล้มเหลว")

    พยายาม:
        Raise_my_Exception()
    ยกเว้น ข้อยกเว้น:
        ผ่าน

    พิมพ์ (raise_my_Exception.statistics)

.. ทดสอบผลลัพธ์::
   :ซ่อน:

   -

การโทรกลับที่กำหนดเอง

คุณยังสามารถกำหนดการโทรกลับของคุณเองได้ การโทรกลับควรยอมรับหนึ่งพารามิเตอร์ที่เรียกว่า retry_state ซึ่งมีข้อมูลทั้งหมดเกี่ยวกับการร้องขอการลองใหม่ในปัจจุบัน

ตัวอย่างเช่น คุณสามารถเรียกใช้ฟังก์ชันการเรียกกลับแบบกำหนดเองได้หลังจากที่ลองใหม่ทั้งหมดล้มเหลว โดยไม่มีข้อยกเว้น (หรือคุณสามารถเพิ่มใหม่หรือทำอะไรก็ได้จริงๆ)

 .. รหัสทดสอบ::

    def return_last_value (ลองอีกครั้ง_state):
        """ส่งคืนผลลัพธ์ของความพยายามโทรครั้งล่าสุด"""
        กลับ retry_state.outcome.result ()

    def is_false (ค่า):
        """คืนค่า True ถ้าค่าเป็นเท็จ"""
        ค่าที่ส่งคืนเป็นเท็จ

    # จะกลับมาเป็นเท็จหลังจากพยายาม 3 ครั้งเพื่อให้ได้ผลลัพธ์ที่แตกต่าง
    @retry(stop=stop_after_attempt(3),
           retry_error_callback=return_last_value,
           ลองอีกครั้ง=retry_if_result(is_false))
    def ในที่สุด_return_false():
        กลับเท็จ

ลอง CallState อีกครั้ง

อาร์กิวเมนต์ retry_state เป็นอ็อบเจ็กต์ของ คลาส :class:`~tenacity.RetryCallState`

การโทรกลับที่กำหนดเองอื่น ๆ

นอกจากนี้ยังสามารถกำหนดการเรียกกลับที่กำหนดเองสำหรับอาร์กิวเมนต์คำหลักอื่นๆ ได้อีกด้วย

 .. ฟังก์ชั่น:: my_stop(retry_state)

   :param RetryCallState retry_state: ข้อมูลเกี่ยวกับการร้องขอการลองใหม่ในปัจจุบัน
   :return: ควรหยุดการลองอีกครั้งหรือไม่
   :rtype: bool

.. ฟังก์ชั่น:: my_wait(retry_state)

   :param RetryCallState retry_state: ข้อมูลเกี่ยวกับการร้องขอการลองใหม่ในปัจจุบัน
   :return: จำนวนวินาทีที่รอก่อนลองอีกครั้งครั้งถัดไป
   :rtype: ลอย

.. ฟังก์ชั่น:: my_retry(retry_state)

   :param RetryCallState retry_state: ข้อมูลเกี่ยวกับการร้องขอการลองใหม่ในปัจจุบัน
   :return: ควรลองใหม่อีกครั้งหรือไม่
   :rtype: bool

.. ฟังก์ชั่น:: 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 (สตรีม = sys.stderr, ระดับ = logging. DEBUG)

    คนตัดไม้ = logging.getLogger(__name__)

    def my_before_sleep (ลองอีกครั้ง_state):
        ถ้า retry_state.attempt_number < 1:
            loglevel = logging.INFO
        อื่น:
            loglevel = การบันทึกคำเตือน
        คนตัดไม้.log(
            loglevel, 'กำลังลอง %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 Raise_my_Exception():
        เพิ่ม MyException("ล้มเหลว")

    พยายาม:
        Raise_my_Exception()
    ยกเว้นข้อผิดพลาดในการลองใหม่:
        ผ่าน

การเปลี่ยนอาร์กิวเมนต์ ณ รันไทม์

คุณสามารถเปลี่ยนอาร์กิวเมนต์ของมัณฑนากรลองใหม่ได้ตามต้องการเมื่อเรียกใช้โดยใช้ฟังก์ชัน retry_with ที่แนบมากับฟังก์ชันที่ห่อ:

 .. รหัสทดสอบ::

    @ลองอีกครั้ง(หยุด=stop_after_attempt(3))
    def Raise_my_Exception():
        เพิ่ม MyException("ล้มเหลว")

    พยายาม:
        Raise_my_Exception.retry_with(หยุด=stop_after_attempt(4))()
    ยกเว้น ข้อยกเว้น:
        ผ่าน

    พิมพ์ (raise_my_Exception.statistics)

.. ทดสอบผลลัพธ์::
   :ซ่อน:

   -

หากคุณต้องการใช้ตัวแปรเพื่อตั้งค่าพารามิเตอร์การลองใหม่ คุณไม่จำเป็นต้องใช้ตัวตกแต่งการลองใหม่ คุณสามารถใช้ Retrying ได้โดยตรงแทน:

 .. รหัสทดสอบ::

    def never_good_enough (arg1):
        เพิ่มข้อยกเว้น ('อาร์กิวเมนต์ไม่ถูกต้อง: {}'.format (arg1))

    def try_never_good_enough(max_attempts=3):
        retryer = ลองใหม่ (หยุด=stop_after_attempt(max_attempts), เพิ่มใหม่=True)
        ลองอีกครั้ง (ไม่เคยดีพอ 'ฉันพยายามจริงๆ')

คุณอาจต้องการเปลี่ยนพฤติกรรมของฟังก์ชันที่ได้รับการตกแต่งชั่วคราว เช่น ในการทดสอบ เพื่อหลีกเลี่ยงเวลารอที่ไม่จำเป็น คุณสามารถแก้ไข/แก้ไขแอตทริบิวต์ลองใหม่ที่แนบมากับฟังก์ชันได้ โปรดทราบว่านี่เป็นแอตทริบิวต์แบบเขียนอย่างเดียว สถิติควรอ่านได้จากแอตทริบิวต์สถิติของฟังก์ชัน

 .. รหัสทดสอบ::

    @retry(stop=stop_after_attempt(3), wait=wait_fixed(3))
    def Raise_my_Exception():
        เพิ่ม MyException("ล้มเหลว")

    จากการจำลองการนำเข้า unittest

    ด้วย mock.patch.object(raise_my_Exception.retry, "รอ", wait_fixed(0)):
        พยายาม:
            Raise_my_Exception()
        ยกเว้น ข้อยกเว้น:
            ผ่าน

    พิมพ์ (raise_my_Exception.statistics)

.. ทดสอบผลลัพธ์::
   :ซ่อน:

   -

กำลังลองบล็อกโค้ดอีกครั้ง

Tenacity ช่วยให้คุณสามารถลองบล็อคโค้ดอีกครั้งโดยไม่จำเป็นต้องล้อมมันไว้ในฟังก์ชันที่แยกออกมา ทำให้ง่ายต่อการแยกบล็อกที่ล้มเหลวขณะแชร์บริบท เคล็ดลับคือการรวม for loop และตัวจัดการบริบทเข้าด้วยกัน

 .. รหัสทดสอบ::

   จากการนำเข้าความดื้อรั้น ลองใหม่, RetryError, stop_after_attempt

   พยายาม:
       สำหรับความพยายามในการลองอีกครั้ง (stop=stop_after_attempt(3)):
           ด้วยความพยายาม:
               เพิ่มข้อยกเว้น ('รหัสของฉันล้มเหลว!')
   ยกเว้นข้อผิดพลาดในการลองใหม่:
       ผ่าน

คุณสามารถกำหนดค่ารายละเอียดทั้งหมดของนโยบายการลองใหม่ได้โดยการกำหนดค่าออบเจ็กต์การลองใหม่

ด้วยโค้ด async คุณสามารถใช้ AsyncRetrying ได้

 .. รหัสทดสอบ::

   จากการนำเข้าความดื้อรั้น AsyncRetrying, RetryError, stop_after_attempt

   ฟังก์ชั่น async def ():
      พยายาม:
          async สำหรับความพยายามใน AsyncRetrying (stop=stop_after_attempt(3)):
              ด้วยความพยายาม:
                  เพิ่มข้อยกเว้น ('รหัสของฉันล้มเหลว!')
      ยกเว้นข้อผิดพลาดในการลองใหม่:
          ผ่าน

ในทั้งสองกรณี คุณอาจต้องการตั้งค่าผลลัพธ์เป็นความพยายามเพื่อให้สามารถใช้ได้ในกลยุทธ์การลองใหม่ เช่น retry_if_result ซึ่งสามารถทำได้โดยการเข้าถึงคุณสมบัติ retry_state :

 .. รหัสทดสอบ::

    จากการนำเข้าความดื้อรั้น AsyncRetrying, retry_if_result

    ฟังก์ชั่น async def ():
       async สำหรับความพยายามใน AsyncRetrying (retry=retry_if_result(lambda x: x < 3)):
           ด้วยความพยายาม:
               result = 1 # การคำนวณที่ซับซ้อน การเรียกใช้ฟังก์ชัน ฯลฯ
           หากไม่ใช่ try.retry_state.outcome.failed:
               พยายาม retry_state.set_result (ผลลัพธ์)
       ส่งคืนผลลัพธ์

Async และลองอีกครั้ง

สุดท้าย retry ครั้งบน coroutines ของ 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 )

คุณยังสามารถใช้ลูปเหตุการณ์ทางเลือก เช่น curio ได้โดยผ่านฟังก์ชันสลีปที่ถูกต้อง:

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

มีส่วนช่วย

1. ตรวจสอบปัญหาที่เปิดอยู่หรือเปิดประเด็นใหม่เพื่อเริ่มการสนทนาเกี่ยวกับแนวคิดเกี่ยวกับฟีเจอร์หรือจุดบกพร่อง
2. แยกพื้นที่เก็บข้อมูลบน GitHub เพื่อเริ่มทำการเปลี่ยนแปลงกับสาขา หลัก (หรือแยกสาขาออกไป)
3. เขียนการทดสอบที่แสดงว่าจุดบกพร่องได้รับการแก้ไขแล้วหรือฟีเจอร์ทำงานได้ตามที่คาดไว้
4. เพิ่มบันทึกการเปลี่ยนแปลง
5. ทำให้เอกสารดีขึ้น (หรือละเอียดมากขึ้น หรืออ่านง่ายขึ้น หรือ ...)

บันทึกการเปลี่ยนแปลง

Reno ใช้สำหรับจัดการบันทึกการเปลี่ยนแปลง ดูเอกสารการใช้งานของพวกเขา

การสร้างเอกสารจะรวบรวมบันทึกการเปลี่ยนแปลงโดยอัตโนมัติ คุณเพียงแค่ต้องเพิ่มพวกเขา

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