tenacity
tenacity 9.0.0
Consulte a documentação de tenacidade para uma melhor experiência.
Tenacity é uma biblioteca de repetição de uso geral licenciada pelo Apache 2.0, escrita em Python, para simplificar a tarefa de adicionar comportamento de repetição a praticamente qualquer coisa. Origina-se de uma bifurcação de novas tentativas que, infelizmente, não é mais mantida. Tenacity não é compatível com API para novas tentativas, mas adiciona novas funcionalidades significativas e corrige vários bugs antigos.
O caso de uso mais simples é tentar novamente uma função instável sempre que ocorre uma exceção até que um valor seja retornado.
.. código de teste::
importar aleatoriamente
da tenacidade, nova tentativa de importação
@tentar novamente
def do_something_unreliable():
se random.randint(0, 10) > 1:
raise IOError("Molho quebrado, está tudo lavado!!!111one")
outro:
retornar "Molho incrível!"
imprimir(do_something_unreliable())
.. saída de teste:: :esconder: Molho incrível!
..toctree::
:escondido:
:profundidade máxima: 2
registro de alterações
API
Para instalar tenacity , simplesmente:
$ pip install tenacity .. configuração de teste::
registro de importação
#
# Observe que a importação a seguir é usada apenas para conveniência de demonstração.
# O código de produção deve sempre importar explicitamente os nomes necessários.
#
da importação de tenacidade *
classe MinhaExceção(Exceção):
passar
Como você viu acima, o comportamento padrão é tentar novamente sem esperar quando uma exceção é gerada.
.. código de teste::
@tentar novamente
def nunca_gonna_give_you_up():
print("Tente novamente ignorando exceções, não espere entre novas tentativas")
levantar exceção
Vamos ser um pouco menos persistentes e estabelecer alguns limites, como o número de tentativas antes de desistir.
.. código de teste::
@retry(stop=stop_after_attempt(7))
def stop_after_7_attempts():
print("Parando após 7 tentativas")
levantar exceção
Não temos o dia todo, então vamos estabelecer um limite para quanto tempo devemos tentar novamente.
.. código de teste::
@retry(stop=stop_after_delay(10))
def stop_after_10_s():
print("Parando após 10 segundos")
levantar exceção
Se você estiver com um prazo apertado e exceder o tempo de atraso não for aceitável, você pode desistir de novas tentativas uma vez antes de exceder o atraso.
.. código de teste::
@retry(stop=stop_before_delay(10))
def stop_before_10_s():
print("Parando 1 tentativa antes de 10 segundos")
levantar exceção
Você pode combinar diversas condições de parada usando o | operador:
.. código de teste::
@retry(stop=(stop_after_delay(10) | stop_after_attempt(5)))
def stop_after_10_s_or_5_retries():
print("Parando após 10 segundos ou 5 tentativas")
levantar exceção
A maioria das coisas não gosta de ser pesquisada o mais rápido possível, então vamos esperar 2 segundos entre as novas tentativas.
.. código de teste::
@retry(wait=wait_fixed(2))
def wait_2_s():
print("Aguarde 2 segundos entre novas tentativas")
levantar exceção
Algumas coisas funcionam melhor com um pouco de aleatoriedade injetada.
.. código de teste::
@retry(wait=wait_random(min=1, max=2))
def wait_random_1_to_2_s():
print("Aguarde aleatoriamente de 1 a 2 segundos entre novas tentativas")
levantar exceção
Por outro lado, é difícil superar a espera exponencial ao tentar novamente serviços distribuídos e outros endpoints remotos.
.. código de teste::
@retry(wait=wait_exponencial(multiplicador=1, min=4, max=10))
def wait_exponencial_1():
print("Aguarde 2 ^ x * 1 segundo entre cada nova tentativa, começando com 4 segundos, depois até 10 segundos e depois 10 segundos depois")
levantar exceção
Por outro lado, também é difícil superar a combinação de esperas fixas e jitter (para ajudar a evitar rebanhos trovejantes) ao tentar novamente serviços distribuídos e outros endpoints remotos.
.. código de teste::
@retry(wait=wait_fixed(3) + wait_random(0, 2))
def wait_fixed_jitter():
print("Aguarde pelo menos 3 segundos e adicione até 2 segundos de atraso aleatório")
levantar exceção
Quando vários processos estão em disputa por um recurso compartilhado, o aumento exponencial do jitter ajuda a minimizar as colisões.
.. código de teste::
@retry(wait=wait_random_exponencial(multiplicador=1, máx=60))
def wait_exponencial_jitter():
print("Aguarde aleatoriamente até 2 ^ x * 1 segundos entre cada nova tentativa até que o intervalo atinja 60 segundos e, em seguida, aleatoriamente até 60 segundos depois")
levantar exceção
Às vezes é necessário construir uma cadeia de recuos.
.. código de teste::
@retry(wait=wait_chain(*[wait_fixed(3) para i no intervalo(3)] +
[wait_fixed(7) para i no intervalo(2)] +
[espera_fixo(9)]))
def wait_fixed_chained():
print("Aguarde 3s para 3 tentativas, 7s para as próximas 2 tentativas e 9s para todas as tentativas seguintes")
levantar exceção
Temos algumas opções para lidar com novas tentativas que levantam exceções específicas ou gerais, como nos casos aqui.
.. código de teste::
classe ClientError (Exceção):
"""Algum tipo de erro do cliente."""
@retry(retry=retry_if_exception_type(IOError))
def pode_io_error():
print("Tente novamente sem esperar se ocorrer um IOError, gere quaisquer outros erros")
levantar exceção
@retry(retry=retry_if_not_exception_type(ClientError))
def pode_cliente_error():
print("Tente novamente para sempre sem esperar se ocorrer algum erro diferente de ClientError. Aumente ClientError imediatamente.")
levantar exceção
Também podemos usar o resultado da função para alterar o comportamento da nova tentativa.
.. código de teste::
def is_none_p(valor):
"""Retorna True se o valor for Nenhum"""
o valor de retorno é Nenhum
@repetir(repetir=repetir_if_result(is_none_p))
def pode_return_none():
print("Tente novamente sem esperar se o valor de retorno for Nenhum")
Veja também estes métodos:
.. código de teste::
retry_if_exception
retry_if_exception_type
retry_if_not_exception_type
retry_unless_exception_type
tentar_se_resultado
retry_if_not_result
retry_if_exception_message
retry_if_not_exception_message
tentar novamente_qualquer
tentar_tudo
Também podemos combinar várias condições:
.. código de teste::
def is_none_p(valor):
"""Retorna True se o valor for Nenhum"""
o valor de retorno é Nenhum
@retry(retry=(retry_if_result(is_none_p) | retry_if_exception_type()))
def pode_return_none():
print("Tente novamente ignorar sempre as exceções sem esperar se o valor de retorno for Nenhum")
Qualquer combinação de parar, esperar, etc. também é suportada para lhe dar a liberdade de misturar e combinar.
Também é possível tentar novamente explicitamente a qualquer momento, gerando a exceção TryAgain:
.. código de teste::
@tentar novamente
def do_alguma coisa():
resultado = algo_outro()
se resultado == 23:
aumentar Tente novamente
Normalmente, quando sua função falha na última vez (e não será tentada novamente com base em suas configurações), um RetryError é gerado. A exceção encontrada pelo seu código será mostrada em algum lugar no meio do rastreamento de pilha.
Se preferir ver a exceção que seu código encontrou no final do rastreamento de pilha (onde é mais visível), você pode definir reraise=True.
.. código de teste::
@retry(reraise=True, stop=stop_after_attempt(3))
def raise_my_exception():
raise MyException("Falha")
tentar:
aumentar_minha_exceção()
exceto MinhaExceção:
# expirou ao tentar novamente
passar
É possível executar uma ação antes de qualquer tentativa de chamar a função usando a função before callback:
.. código de teste::
registro de importação
sistema de importação
logging.basicConfig(stream=sys.stderr, nível=logging.DEBUG)
registrador = logging.getLogger(__name__)
@retry(stop=stop_after_attempt(3), before=before_log(logger, logging.DEBUG))
def raise_my_exception():
raise MyException("Falha")
No mesmo espírito, é possível executar após uma chamada que falhou:
.. código de teste::
registro de importação
sistema de importação
logging.basicConfig(stream=sys.stderr, nível=logging.DEBUG)
registrador = logging.getLogger(__name__)
@retry(stop=stop_after_attempt(3), after=after_log(logger, logging.DEBUG))
def raise_my_exception():
raise MyException("Falha")
Também é possível registrar apenas falhas que serão repetidas. Normalmente as novas tentativas acontecem após um intervalo de espera, então o argumento da palavra-chave é chamado before_sleep :
.. código de teste::
registro de importação
sistema de importação
logging.basicConfig(stream=sys.stderr, nível=logging.DEBUG)
registrador = logging.getLogger(__name__)
@retry(stop=stop_after_attempt(3),
before_sleep=before_sleep_log(logger, logging.DEBUG))
def raise_my_exception():
raise MyException("Falha")
Você pode acessar as estatísticas sobre a nova tentativa feita em uma função usando o atributo Statistics anexado à função:
.. código de teste::
@retry(stop=stop_after_attempt(3))
def raise_my_exception():
raise MyException("Falha")
tentar:
aumentar_minha_exceção()
exceto Exceção:
passar
imprimir(raise_my_exception.statistics)
.. saída de teste:: :esconder: ...
Você também pode definir seus próprios retornos de chamada. O retorno de chamada deve aceitar um parâmetro chamado retry_state que contém todas as informações sobre a invocação de nova tentativa atual.
Por exemplo, você pode chamar uma função de retorno de chamada personalizada depois que todas as tentativas falharem, sem gerar uma exceção (ou você pode aumentar novamente ou fazer qualquer coisa realmente)
.. código de teste::
def return_last_value(retry_state):
"""retorna o resultado da última tentativa de chamada"""
retornar retry_state.outcome.result()
def é_falso(valor):
"""Retorna True se o valor for False"""
o valor de retorno é falso
# retornará False após tentar 3 vezes obter um resultado diferente
@retry(stop=stop_after_attempt(3),
retry_error_callback=return_last_value,
tentar novamente = tentar_se_resultado (é_falso))
def eventualmente_return_false():
retornar falso
O argumento retry_state é um objeto da classe :class:`~tenacity.RetryCallState` .
Também é possível definir retornos de chamada personalizados para outros argumentos de palavras-chave.
.. função:: my_stop(retry_state) :param RetryCallState retry_state: informações sobre a invocação de nova tentativa atual :return: se a nova tentativa deve ou não parar :rtype: bool
.. função:: my_wait(retry_state) :param RetryCallState retry_state: informações sobre a invocação de nova tentativa atual :return: número de segundos de espera antes da próxima tentativa :rtype: float
.. função:: my_retry(retry_state) :param RetryCallState retry_state: informações sobre a invocação de nova tentativa atual :return: se a nova tentativa deve ou não continuar :rtype: bool
.. função:: my_before(retry_state) :param RetryCallState retry_state: informações sobre a invocação de nova tentativa atual
.. função:: my_after(retry_state) :param RetryCallState retry_state: informações sobre a invocação de nova tentativa atual
.. função:: my_before_sleep(retry_state) :param RetryCallState retry_state: informações sobre a invocação de nova tentativa atual
Aqui está um exemplo com uma função before_sleep personalizada:
.. código de teste::
registro de importação
logging.basicConfig(stream=sys.stderr, nível=logging.DEBUG)
registrador = logging.getLogger(__name__)
def meu_antes_sleep(retry_state):
se retry_state.attempt_number < 1:
nível de log = registro.INFO
outro:
loglevel = logging.WARNING
registrador.log(
loglevel, 'Retentando %s: tentativa %s terminou com: %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("Falha")
tentar:
aumentar_minha_exceção()
exceto RetryError:
passar
Você pode alterar os argumentos de um decorador de repetição conforme necessário ao chamá-lo usando a função retry_with anexada à função encapsulada:
.. código de teste::
@retry(stop=stop_after_attempt(3))
def raise_my_exception():
raise MyException("Falha")
tentar:
raise_my_exception.retry_with(stop=stop_after_attempt(4))()
exceto Exceção:
passar
imprimir(raise_my_exception.statistics)
.. saída de teste:: :esconder: ...
Se quiser usar variáveis para configurar os parâmetros de nova tentativa, você não precisa usar o decorador de nova tentativa - em vez disso, você pode usar Retrying diretamente:
.. código de teste::
def nunca_bom_enough(arg1):
raise Exception('Argumento inválido: {}'.format(arg1))
def try_never_good_enough(max_attempts=3):
retryer = Tentando novamente(stop=stop_after_attempt(max_attempts), reraise=True)
retryer(never_good_enough, 'Eu realmente tento')
Você também pode querer alterar temporariamente o comportamento de uma função decorada, como em testes, para evitar tempos de espera desnecessários. Você pode modificar/corrigir o atributo de nova tentativa anexado à função. Tenha em mente que este é um atributo somente gravação, as estatísticas devem ser lidas no atributo de estatísticas da função.
.. código de teste::
@retry(stop=stop_after_attempt(3), wait=wait_fixed(3))
def raise_my_exception():
raise MyException("Falha")
da simulação de importação unittest
com mock.patch.object(raise_my_exception.retry, "wait", wait_fixed(0)):
tentar:
aumentar_minha_exceção()
exceto Exceção:
passar
imprimir(raise_my_exception.statistics)
.. saída de teste:: :esconder: ...
O Tenacity permite que você tente novamente um bloco de código sem a necessidade de envolvê-lo em uma função isolada. Isso facilita o isolamento de blocos com falha durante o compartilhamento do contexto. O truque é combinar um loop for e um gerenciador de contexto.
.. código de teste::
da importação de tenacidade Retrying, RetryError, stop_after_attempt
tentar:
para tentativa de nova tentativa (stop=stop_after_attempt(3)):
com tentativa:
raise Exception('Meu código está falhando!')
exceto RetryError:
passar
Você pode configurar todos os detalhes da política de nova tentativa configurando o objeto Retrying.
Com código assíncrono você pode usar AsyncRetrying.
.. código de teste::
da importação de tenacidade AsyncRetrying, RetryError, stop_after_attempt
função def assíncrona():
tentar:
assíncrono para tentativa em AsyncRetrying(stop=stop_after_attempt(3)):
com tentativa:
raise Exception('Meu código está falhando!')
exceto RetryError:
passar
Em ambos os casos, convém definir o resultado da tentativa para que fique disponível em estratégias de repetição, como retry_if_result . Isso pode ser feito acessando a propriedade retry_state :
.. código de teste::
da importação de tenacidade AsyncRetrying, retry_if_result
função def assíncrona():
assíncrono para tentativa em AsyncRetrying(retry=retry_if_result(lambda x: x < 3)):
com tentativa:
resultado = 1 # Alguns cálculos complexos, chamadas de função, etc.
se não, try.retry_state.outcome.failed:
tentativa.retry_state.set_result(resultado)
resultado de retorno
Por fim, retry também funciona em corrotinas asyncio, Trio e Tornado (>= 4,5). As suspensões também são feitas de forma assíncrona.
@ 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 )Você pode até usar loops de eventos alternativos, como curio, passando a função sleep correta:
@ retry ( sleep = curio . sleep )
async def my_async_curio_function ():
await asks . get ( 'https://example.org' )reno é usado para gerenciar changelogs. Dê uma olhada em seus documentos de uso.
A geração do documento compilará automaticamente os changelogs. Você só precisa adicioná-los.
# Opens a template file in an editor
tox -e reno -- new some-slug-for-my-change --edit
