tenacity
tenacity 9.0.0
Für eine bessere Erfahrung lesen Sie bitte die Tenacity-Dokumentation .
Tenacity ist eine von Apache 2.0 lizenzierte Allzweck-Wiederholungsbibliothek, die in Python geschrieben wurde, um das Hinzufügen von Wiederholungsverhalten zu nahezu allem zu vereinfachen. Der Ursprung liegt in einem Retry-Fork, der leider nicht mehr gepflegt wird. Tenacity ist nicht API-kompatibel mit Wiederholungsversuchen, fügt jedoch wichtige neue Funktionen hinzu und behebt eine Reihe langjähriger Fehler.
Der einfachste Anwendungsfall besteht darin, eine Flaky-Funktion immer dann zu wiederholen, wenn eine Ausnahme auftritt, bis ein Wert zurückgegeben wird.
.. Testcode::
Zufällig importieren
from tenacity import retry
@wiederholen
def do_something_unreliable():
wenn random.randint(0, 10) > 1:
raise IOError("Kaputte Soße, alles ist abgespritzt!!!111one")
anders:
zurück: „Tolle Soße!“
print(do_something_unreliable())
.. testausgabe:: :verstecken: Tolle Soße!
.. toctree::
:versteckt:
:maxtiefe: 2
Änderungsprotokoll
API
Um tenacity zu installieren, gehen Sie einfach wie folgt vor:
$ pip install tenacity .. Testaufbau::
Protokollierung importieren
#
# Beachten Sie, dass der folgende Import nur zu Demonstrationszwecken verwendet wird.
# Produktionscode sollte die benötigten Namen immer explizit importieren.
#
aus Tenacity-Import *
Klasse MyException(Exception):
passieren
Wie Sie oben gesehen haben, besteht das Standardverhalten darin, es immer wieder zu versuchen, ohne zu warten, wenn eine Ausnahme ausgelöst wird.
.. Testcode::
@wiederholen
def never_gonna_give_you_up():
print("Immer wieder versuchen, Ausnahmen ignorieren, nicht zwischen den Wiederholungsversuchen warten")
Ausnahme auslösen
Seien wir etwas weniger hartnäckig und setzen wir ein paar Grenzen, etwa die Anzahl der Versuche, bevor wir aufgeben.
.. Testcode::
@retry(stop=stop_after_attempt(7))
def stop_after_7_attempts():
print("Stoppt nach 7 Versuchen")
Ausnahme auslösen
Wir haben nicht den ganzen Tag Zeit, also legen wir eine Grenze dafür fest, wie lange wir etwas wiederholen sollten.
.. Testcode::
@retry(stop=stop_after_delay(10))
def stop_after_10_s():
print("Stoppt nach 10 Sekunden")
Ausnahme auslösen
Wenn Ihre Frist knapp ist und eine Überschreitung der Verzögerungszeit nicht in Ordnung ist, können Sie auf Wiederholungsversuche einen Versuch verzichten, bevor die Verzögerung überschritten wird.
.. Testcode::
@retry(stop=stop_before_delay(10))
def stop_before_10_s():
print("1 Versuch vor 10 Sekunden stoppen")
Ausnahme auslösen
Sie können mehrere Stoppbedingungen kombinieren, indem Sie das | verwenden Operator:
.. Testcode::
@retry(stop=(stop_after_delay(10) | stop_after_attempt(5)))
def stop_after_10_s_or_5_retries():
print("Stoppt nach 10 Sekunden oder 5 Wiederholungsversuchen")
Ausnahme auslösen
Die meisten Dinge mögen es nicht, so schnell wie möglich abgefragt zu werden, also warten wir einfach zwei Sekunden zwischen den Wiederholungsversuchen.
.. Testcode::
@retry(wait=wait_fixed(2))
def wait_2_s():
print("Warten Sie 2 Sekunden zwischen den Wiederholungsversuchen")
Ausnahme auslösen
Manche Dinge funktionieren am besten, wenn ein wenig Zufälligkeit hinzugefügt wird.
.. Testcode::
@retry(wait=wait_random(min=1, max=2))
def wait_random_1_to_2_s():
print("Warten Sie nach dem Zufallsprinzip 1 bis 2 Sekunden zwischen den Wiederholungsversuchen")
Ausnahme auslösen
Andererseits ist es schwierig, den exponentiellen Backoff zu überwinden, wenn verteilte Dienste und andere Remote-Endpunkte erneut versucht werden.
.. Testcode::
@retry(wait=wait_exponential(multiplier=1, min=4, max=10))
def wait_exponential_1():
print("Warten Sie 2^x * 1 Sekunde zwischen jedem Wiederholungsversuch, beginnend mit 4 Sekunden, dann bis zu 10 Sekunden, dann 10 Sekunden danach")
Ausnahme auslösen
Andererseits ist es auch schwer, feste Wartezeiten und Jitter zu kombinieren (um donnernde Herden zu vermeiden), wenn verteilte Dienste und andere Remote-Endpunkte erneut versucht werden.
.. Testcode::
@retry(wait=wait_fixed(3) + wait_random(0, 2))
def wait_fixed_jitter():
print("Warten Sie mindestens 3 Sekunden und fügen Sie bis zu 2 Sekunden zufällige Verzögerung hinzu")
Ausnahme auslösen
Wenn mehrere Prozesse um eine gemeinsame Ressource konkurrieren, trägt ein exponentiell steigender Jitter dazu bei, Kollisionen zu minimieren.
.. Testcode::
@retry(wait=wait_random_exponential(multiplier=1, max=60))
def wait_exponential_jitter():
print("Warten Sie nach dem Zufallsprinzip bis zu 2^x * 1 Sekunden zwischen jedem Wiederholungsversuch, bis der Bereich 60 Sekunden erreicht, und danach bis zu 60 Sekunden nach dem Zufallsprinzip.")
Ausnahme auslösen
Manchmal ist es notwendig, eine Backoff-Kette aufzubauen.
.. 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("Warten Sie 3 Sekunden bei 3 Versuchen, 7 Sekunden bei den nächsten 2 Versuchen und 9 Sekunden bei allen weiteren Versuchen danach")
Ausnahme auslösen
Wir haben einige Möglichkeiten, mit Wiederholungsversuchen umzugehen, die bestimmte oder allgemeine Ausnahmen auslösen, wie in den Fällen hier.
.. Testcode::
Klasse ClientError(Exception):
„“„Irgendein Client-Fehler.“““
@retry(retry=retry_if_Exception_type(IOError))
def might_io_error():
print("Immer ohne Wartezeit erneut versuchen, wenn ein IOError auftritt, alle anderen Fehler auslösen")
Ausnahme auslösen
@retry(retry=retry_if_not_Exception_type(ClientError))
def might_client_error():
print("Wenn ein anderer Fehler als ClientError auftritt, wiederholen Sie den Vorgang ohne Wartezeit. Lösen Sie ClientError sofort aus.")
Ausnahme auslösen
Wir können das Ergebnis der Funktion auch verwenden, um das Verhalten bei Wiederholungsversuchen zu ändern.
.. Testcode::
def is_none_p(value):
„“„Gib True zurück, wenn der Wert None ist““
Rückgabewert ist None
@retry(retry=retry_if_result(is_none_p))
def might_return_none():
print("Wiederholen ohne Wartezeit, wenn der Rückgabewert None ist")
Siehe auch diese Methoden:
.. Testcode::
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
retry_any
retry_all
Wir können auch mehrere Bedingungen kombinieren:
.. Testcode::
def is_none_p(value):
„“„Gib True zurück, wenn der Wert None ist““
Rückgabewert ist None
@retry(retry=(retry_if_result(is_none_p) | retry_if_Exception_type()))
def might_return_none():
print("Wiederholen Sie den Versuch, Ausnahmen ohne Wartezeit zu ignorieren, wenn der Rückgabewert None ist")
Jede beliebige Kombination aus Stopp, Warten usw. wird ebenfalls unterstützt, um Ihnen die Freiheit zu geben, sie zu kombinieren.
Es ist auch jederzeit möglich, es explizit noch einmal zu versuchen, indem Sie die TryAgain-Ausnahme auslösen:
.. Testcode::
@wiederholen
def do_something():
result = Something_else()
wenn Ergebnis == 23:
Erhöhen Sie TryAgain
Normalerweise wird ein RetryError ausgelöst, wenn Ihre Funktion beim letzten Mal fehlschlägt (und basierend auf Ihren Einstellungen nicht noch einmal versucht wird). Die Ausnahme, auf die Ihr Code gestoßen ist, wird irgendwo in der Mitte des Stack-Trace angezeigt.
Wenn Sie die Ausnahme, auf die Ihr Code gestoßen ist, lieber am Ende des Stack-Trace sehen möchten (dort, wo sie am sichtbarsten ist), können Sie reraise=True festlegen.
.. Testcode::
@retry(reraise=True, stop=stop_after_attempt(3))
def raise_my_Exception():
raise MyException("Fail")
versuchen:
raise_my_Exception()
außer MyException:
# Zeitüberschreitung beim erneuten Versuch
passieren
Es ist möglich, eine Aktion vor jedem Versuch, die Funktion aufzurufen, auszuführen, indem Sie die Funktion „before callback“ verwenden:
.. Testcode::
Protokollierung importieren
Importsystem
logging.basicConfig(stream=sys.stderr, level=logging.DEBUG)
logger = logging.getLogger(__name__)
@retry(stop=stop_after_attempt(3), before=before_log(logger, logging.DEBUG))
def raise_my_Exception():
raise MyException("Fail")
Im gleichen Sinne ist es möglich, nach einem fehlgeschlagenen Anruf Folgendes auszuführen:
.. Testcode::
Protokollierung importieren
Importsystem
logging.basicConfig(stream=sys.stderr, level=logging.DEBUG)
logger = logging.getLogger(__name__)
@retry(stop=stop_after_attempt(3), after=after_log(logger, logging.DEBUG))
def raise_my_Exception():
raise MyException("Fail")
Es ist auch möglich, nur Fehler zu protokollieren, die erneut versucht werden. Normalerweise finden Wiederholungsversuche nach einem Warteintervall statt, daher heißt das Schlüsselwortargument before_sleep :
.. Testcode::
Protokollierung importieren
Importsystem
logging.basicConfig(stream=sys.stderr, level=logging.DEBUG)
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")
Sie können auf die Statistiken über die über eine Funktion durchgeführten Wiederholungsversuche zugreifen, indem Sie das an die Funktion angehängte Statistikattribut verwenden:
.. Testcode::
@retry(stop=stop_after_attempt(3))
def raise_my_Exception():
raise MyException("Fail")
versuchen:
raise_my_Exception()
außer Ausnahme:
passieren
print(raise_my_Exception.statistics)
.. testausgabe:: :verstecken: ...
Sie können auch Ihre eigenen Rückrufe definieren. Der Rückruf sollte einen Parameter namens retry_state akzeptieren, der alle Informationen zum aktuellen Wiederholungsaufruf enthält.
Sie können beispielsweise eine benutzerdefinierte Rückruffunktion aufrufen, nachdem alle Wiederholungsversuche fehlgeschlagen sind, ohne eine Ausnahme auszulösen (oder Sie können eine Ausnahme erneut auslösen oder wirklich etwas anderes tun).
.. Testcode::
def return_last_value(retry_state):
„““gibt das Ergebnis des letzten Anrufversuchs zurück““
return retry_state.outcome.result()
def is_false(value):
"""Gib True zurück, wenn der Wert False ist""
Rückgabewert ist False
# gibt False zurück, nachdem dreimal versucht wurde, ein anderes Ergebnis zu erhalten
@retry(stop=stop_after_attempt(3),
retry_error_callback=return_last_value,
retry=retry_if_result(is_false))
def evenual_return_false():
Gibt False zurück
Das Argument retry_state ist ein Objekt der Klasse :class:`~tenacity.RetryCallState` .
Es ist auch möglich, benutzerdefinierte Rückrufe für andere Schlüsselwortargumente zu definieren.
.. Funktion:: my_stop(retry_state) :param RetryCallState retry_state: Informationen zum aktuellen Wiederholungsaufruf :return: Gibt an, ob der erneute Versuch gestoppt werden soll oder nicht :rtype: bool
.. Funktion:: my_wait(retry_state) :param RetryCallState retry_state: Informationen zum aktuellen Wiederholungsaufruf :return: Anzahl der Sekunden, die vor dem nächsten Wiederholungsversuch gewartet werden soll :rtype: float
.. Funktion:: my_retry(retry_state) :param RetryCallState retry_state: Informationen zum aktuellen Wiederholungsaufruf :return: Gibt an, ob der erneute Versuch fortgesetzt werden soll oder nicht :rtype: bool
.. function:: my_before(retry_state) :param RetryCallState retry_state: Informationen zum aktuellen Wiederholungsaufruf
.. function:: my_after(retry_state) :param RetryCallState retry_state: Informationen zum aktuellen Wiederholungsaufruf
.. Funktion:: my_before_sleep(retry_state) :param RetryCallState retry_state: Informationen zum aktuellen Wiederholungsaufruf
Hier ist ein Beispiel mit einer benutzerdefinierten before_sleep -Funktion:
.. Testcode::
Protokollierung importieren
logging.basicConfig(stream=sys.stderr, level=logging.DEBUG)
logger = logging.getLogger(__name__)
def my_before_sleep(retry_state):
wenn retry_state.attempt_number < 1:
loglevel = logging.INFO
anders:
loglevel = logging.WARNUNG
logger.log(
loglevel, 'Wiederholungsversuch %s: Versuch %s endete mit: %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():
raise MyException("Fail")
versuchen:
raise_my_Exception()
außer RetryError:
passieren
Sie können die Argumente eines Retry-Dekorators nach Bedarf ändern, wenn Sie ihn aufrufen, indem Sie die Funktion retry_with verwenden, die an die umschlossene Funktion angehängt ist:
.. Testcode::
@retry(stop=stop_after_attempt(3))
def raise_my_Exception():
raise MyException("Fail")
versuchen:
raise_my_Exception.retry_with(stop=stop_after_attempt(4))()
außer Ausnahme:
passieren
print(raise_my_Exception.statistics)
.. testausgabe:: :verstecken: ...
Wenn Sie Variablen verwenden möchten, um die Wiederholungsparameter einzurichten, müssen Sie nicht den Wiederholungsdekorator verwenden – Sie können stattdessen „Wiederholung“ direkt verwenden:
.. Testcode::
def never_good_enough(arg1):
raise Exception('Ungültiges Argument: {}'.format(arg1))
def try_never_good_enough(max_attempts=3):
retryer = Wiederholen (stop=stop_after_attempt(max_attempts), reraise=True)
retryer(never_good_enough, 'Ich versuche es wirklich')
Möglicherweise möchten Sie auch das Verhalten einer dekorierten Funktion vorübergehend ändern, z. B. bei Tests, um unnötige Wartezeiten zu vermeiden. Sie können das an die Funktion angehängte Wiederholungsattribut ändern/patchen. Beachten Sie, dass es sich hierbei um ein schreibgeschütztes Attribut handelt. Statistiken sollten aus dem Funktionsstatistikattribut gelesen werden.
.. Testcode::
@retry(stop=stop_after_attempt(3), wait=wait_fixed(3))
def raise_my_Exception():
raise MyException("Fail")
von Unittest Import Mock
mit mock.patch.object(raise_my_exclusion.retry, "wait", wait_fixed(0)):
versuchen:
raise_my_Exception()
außer Ausnahme:
passieren
print(raise_my_Exception.statistics)
.. testausgabe:: :verstecken: ...
Mit Tenacity können Sie einen Codeblock wiederholen, ohne ihn in eine isolierte Funktion einschließen zu müssen. Dies macht es einfach, fehlerhafte Blöcke zu isolieren und gleichzeitig den Kontext zu teilen. Der Trick besteht darin, eine for-Schleife und einen Kontextmanager zu kombinieren.
.. Testcode::
from tenacity import Retrying, RetryError, stop_after_attempt
versuchen:
für Versuch in Retrying(stop=stop_after_attempt(3)):
mit Versuch:
Exception auslösen („Mein Code schlägt fehl!“)
außer RetryError:
passieren
Sie können alle Details der Wiederholungsrichtlinie konfigurieren, indem Sie das Wiederholungsobjekt konfigurieren.
Mit asynchronem Code können Sie AsyncRetrying verwenden.
.. Testcode::
aus Tenacity-Import AsyncRetrying, RetryError, stop_after_attempt
asynchrone Def-Funktion ():
versuchen:
async für Versuch in AsyncRetrying(stop=stop_after_attempt(3)):
mit Versuch:
Exception auslösen („Mein Code schlägt fehl!“)
außer RetryError:
passieren
In beiden Fällen möchten Sie möglicherweise das Ergebnis auf den Versuch festlegen, damit es in Wiederholungsstrategien wie retry_if_result verfügbar ist. Dies kann durch Zugriff auf die Eigenschaft retry_state erfolgen:
.. Testcode::
aus Tenacity Import AsyncRetrying, retry_if_result
asynchrone Def-Funktion():
async für Versuch in AsyncRetrying(retry=retry_if_result(lambda x: x < 3)):
mit Versuch:
result = 1 # Einige komplexe Berechnungen, Funktionsaufrufe usw.
Wenn nicht, try.retry_state.outcome.failed:
try.retry_state.set_result(result)
Ergebnis zurückgeben
Schließlich funktioniert retry auch bei Asyncio-, Trio- und Tornado-Coroutinen (>= 4.5). Der Schlaf erfolgt auch asynchron.
@ 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 )Sie können sogar alternative Ereignisschleifen wie Curio verwenden, indem Sie die richtige Schlaffunktion übergeben:
@ retry ( sleep = curio . sleep )
async def my_async_curio_function ():
await asks . get ( 'https://example.org' )reno wird zum Verwalten von Änderungsprotokollen verwendet. Schauen Sie sich ihre Nutzungsdokumente an.
Bei der Dokumentenerstellung werden die Änderungsprotokolle automatisch kompiliert. Sie müssen sie nur hinzufügen.
# Opens a template file in an editor
tox -e reno -- new some-slug-for-my-change --edit
