A automação de reitnentos para operações suscetíveis a falhas transitórias é uma prática essencial no desenvolvimento de software robusto. Em Python, a biblioteca de terceiros retrying oferece uma maneira declarativa e flexível de implementar essa lógica. Este artigo aborda a estrutura e o funcionamento interno desta biblioteca, revelando como ela simplifica a gestão de tentativas repetidas de execução de código.

Para começar a utilizar a biblioteca retrying, a instalação é direta:

pip install retrying

A seguir, são apresentados diversos exemplos práticos que ilustram as capacidades de configuração do decorador @retry:

Exemplo 1: Reintento Contínuo

Configura uma função para ser reintentada indefinidamente em caso de falha, sem qualquer atraso entre as tentativas.

from retrying import retry

@retry
def operacao_infinitas_tentativas():
    """Tenta executar esta função repetidamente sem atrasos."""
    print("Tentando continuamente até ter sucesso...")
    # Lógica que pode falhar

Exemplo 2: Limite de Tentativas

Define um número máximo de tentativas antes de desistir. Neste caso, a função será reintentada no máximo sete vezes.

from retrying import retry

@retry(stop_max_attempt_number=7)
def parar_apos_sete_tentativas():
    """Executa a função, parando após a sétima tentativa."""
    print("Tentativa limitada a sete chamadas.")
    # Lógica que pode falhar

Exemplo 3: Limite de Tempo Total

Especifica um tempo máximo total em milissegundos para todas as tentativas. A reexecução será interrompida após 10 segundos.

from retrying import retry

@retry(stop_max_delay=10000)
def parar_apos_dez_segundos():
    """Para de tentar após dez segundos."""
    print("O ciclo de reintento durará no máximo dez segundos.")
    # Lógica que pode falhar

Exemplo 4: Atraso Fixo entre Tentativas

Determina um período de espera fixo entre cada tentativa falha, aqui configurado para 2 segundos.

from retrying import retry

@retry(wait_fixed=2000)
def esperar_dois_segundos():
    """Aguarde dois segundos entre cada reintento."""
    print("Esperando 2 segundos antes de reintentar.")
    # Lógica que pode falhar

Exemplo 5: Atraso Aleatório

Introduz um atraso aleatório entre tentativas, útil para evitar sobrecarregar um recurso externo com reintentos simultâneos.

from retrying import retry

@retry(wait_random_min=1000, wait_random_max=2000)
def esperar_aleatorio_um_a_dois_segundos():
    """Aguarde entre 1 e 2 segundos aleatoriamente."""
    print("Espera aleatória de 1 a 2 segundos antes de reintentar.")
    # Lógica que pode falhar

Exemplo 6: Espera Exponencial

Implementa um atraso exponencial crescente entre as tentativas, com um multiplicador e um limite máximo.

from retrying import retry

@retry(wait_exponential_multiplier=1000, wait_exponential_max=10000)
def atraso_exponencial():
    """Atraso exponencial: 2^n * 1000ms, limitado a 10 segundos."""
    print("Atraso exponencial entre tentativas, com teto de 10 segundos.")
    # Lógica que pode falhar

Exemplo 7: Reintento Condicional por Exceção

Permite especificar quais tipos de exceções devem acionar um reintento. Outras exceções são propagadas.

from retrying import retry
import io

def re_tentar_se_io_erro(excecao):
    """Retorna True para reintentar se for um IOError, False caso contrário."""
    return isinstance(excecao, io.IOError)

@retry(retry_on_exception=re_tentar_se_io_erro)
def operacao_pode_dar_io_erro():
    """Reintenta apenas se um IOError for levantado."""
    print("Tentando, reintentando apenas em caso de IOError.")
    # Lógica que pode levantar IOError
    raise io.IOError("Simulando um erro de I/O")

@retry(retry_on_exception=re_tentar_se_io_erro, wrap_exception=True)
def levantar_erro_retry_se_nao_io():
    """Reintenta em IOError, encapsula outras exceções em RetryError."""
    print("Tentando, outras exceções são encapsuladas em RetryError.")
    # Lógica que pode levantar outras exceções
    raise ValueError("Simulando outro tipo de erro")

Exemplo 8: Reintento Condicional por Resultado da Função

Define uma condição baseada no valor de retorno da função para determinar se um reintento é necessário.

from retrying import retry

def re_tentar_se_resultado_nulo(resultado):
    """Retorna True para reintentar se o resultado for None, False caso contrário."""
    return resultado is None

@retry(retry_on_result=re_tentar_se_resultado_nulo)
def operacao_pode_retornar_nulo():
    """Reintenta se a função retornar None."""
    print("Tentando, reintentando se o resultado for None.")
    # Lógica que pode retornar None
    import random
    if random.random() < 0.5:
        return None
    return "Sucesso!"

Os exemplos acima demonstram a facilidade de uso do decorador @retry. Mas como ele funciona internamente para fornecer essa funcionalidade não-intrusiva?

O Funcionamento Interno do Decorador @retry

O decorador @retry é uma implementação do padrão de decorador em Python. Ele intercepta a definição da função e a envolve com a lógica de reintento. Existem duas formas principais de usá-lo: @retry (sem parênteses) e @retry(...) (com parênteses e possíveis argumentos).

Considere a implementação simplificada do decorador retry:

import six
import time
import random
import sys
import traceback

# Importação de classes e exceções fictícias para o contexto de análise
class RetryError(Exception):
    def __init__(self, attempt):
        super().__init__(f"Falha na execução após {attempt.attempt_number} tentativas.")
        self.last_attempt = attempt

# A classe Retrying será detalhada mais tarde, mas é essencial para o decorador.
class Retrying:
    def __init__(self, *dargs, **dkw):
        # Simplificado para a análise do decorador.
        # Em uma implementação real, dargs e dkw seriam usados para configurar
        # as estratégias de parada, espera, etc.
        self._config = dkw
        self._config_args = dargs

    def call(self, func_alvo, *args, **kwargs):
        # Esta é a parte central onde a lógica de reintento real aconteceria.
        # Por agora, apenas chamamos a função alvo diretamente para ilustrar o fluxo do decorador.
        print(f"Chamando o método 'call' da instância Retrying para {func_alvo.__name__} com config: {self._config}")
        return func_alvo(*args, **kwargs)

def retry(*decorator_args, **decorator_kwargs):
    """
    Decorador que configura um objeto Retrying e delega a ele a execução da função.
    Suporta @retry e @retry().
    """
    # Caso 1: Uso como @retry (sem parênteses)
    # Neste cenário, decorator_args[0] é a função que está sendo decorada.
    if len(decorator_args) == 1 and callable(decorator_args[0]):
        def envolver_simples(funcao_original):
            @six.wraps(funcao_original)
            def funcao_envolta_simples(*args, **kwargs):
                # Cria uma instância padrão de Retrying sem argumentos
                print(f"DEBUG: @retry (sem args) - Chamando {funcao_original.__name__}")
                return Retrying().call(funcao_original, *args, **kwargs)
            return funcao_envolta_simples
        return envolver_simples(decorator_args[0])
    # Caso 2: Uso como @retry(...) (com parênteses e/ou argumentos)
    # Neste cenário, decorator_args e decorator_kwargs contêm os argumentos de configuração.
    else:
        def envolver_com_config(funcao_original):
            @six.wraps(funcao_original)
            def funcao_envolta_config(*args, **kwargs):
                # Cria uma instância de Retrying com os argumentos passados ao decorador
                print(f"DEBUG: @retry(args) - Chamando {funcao_original.__name__} com args de config: {decorator_kwargs}")
                return Retrying(*decorator_args, **decorator_kwargs).call(funcao_original, *args, **kwargs)
            return funcao_envolta_config
        return envolver_com_config

Quando uma função é decorada com @retry (sem argumentos), como em @retry def minha_funcao(): ..., o Python a interpreta como minha_funcao = retry(minha_funcao). Isso aciona o "Caso 1" no código do decorador, onde a função minha_funcao é passada diretamente para retry. O decorador retorna então uma nova função (funcao_envolta_simples) que, quando chamada, cria uma instância padrão de Retrying e usa seu método call para executar a função original.

Por outro lado, quando usado com argumentos, como em @retry(stop_max_attempt_number=7) def outra_funcao(): ..., o Python executa retry(stop_max_attempt_number=7) primeiro. Isso aciona o "Caso 2", que retorna uma função (envolver_com_config). Essa função é então chamada com outra_funcao como argumento: outra_funcao = envolver_com_config(outra_funcao). O resultado final é uma funcao_envolta_config que, ao ser executada, instancia Retrying com os argumentos fornecidos (stop_max_attempt_number=7, etc.) e chama seu método call.

Em ambos os casos, a chamada da função original é encapsulada dentro do método call de uma instância de Retrying, que é onde a lógica central de reintento reside.

Análise da Classe Retrying e o Método call

O método call dentro da classe Retrying é o motor da biblioteca. Ele orquestra o ciclo de tentativas, verificando condições de parada, aplicando atrasos e lidando com exceções e resultados. A seguir, uma representação do fluxo principal:

class Attempt:
    """
    Uma Attempt encapsula o resultado de uma chamada de função:
    seja um valor retornado ou uma exceção.
    """
    def __init__(self, valor, num_tentativa, tem_excecao):
        self.valor = valor # O resultado ou sys.exc_info() se for exceção
        self.attempt_number = num_tentativa
        self.has_exception = tem_excecao

    def get(self, encapsular_excecao=False):
        """
        Retorna o valor de sucesso ou levanta a exceção capturada.
        Se encapsular_excecao for True, a exceção é envolvida em RetryError.
        """
        if self.has_exception:
            if encapsular_excecao:
                raise RetryError(self)
            else:
                # Re-levanta a exceção original com seu traceback original
                six.reraise(self.valor[0], self.valor[1], self.valor[2])
        else:
            return self.valor

    def __repr__(self):
        if self.has_exception:
            return f"Tentativas: {self.attempt_number}, Erro:\n{''.join(traceback.format_tb(self.valor[2]))}"
        else:
            return f"Tentativas: {self.attempt_number}, Valor: {self.valor}"


class Retrying:
    # ... (Parte do __init__ com configurações de parada e espera) ...
    def __init__(self, *args, **kwargs):
        self._config = kwargs
        # Configuração padrão ou com base nos kwargs
        self._stop_max_attempt_number = kwargs.get('stop_max_attempt_number')
        self._stop_max_delay = kwargs.get('stop_max_delay')
        self._wait_fixed = kwargs.get('wait_fixed')
        self._wait_random_min = kwargs.get('wait_random_min')
        self._wait_random_max = kwargs.get('wait_random_max')
        self._wait_exponential_multiplier = kwargs.get('wait_exponential_multiplier')
        self._wait_exponential_max = kwargs.get('wait_exponential_max')
        self._retry_on_exception = kwargs.get('retry_on_exception', lambda e: True) # Reintenta em qualquer exceção por padrão
        self._retry_on_result = kwargs.get('retry_on_result', lambda r: False) # Não reintenta por resultado por padrão
        self._wrap_exception = kwargs.get('wrap_exception', False)
        self._before_attempts = kwargs.get('before_attempts')
        self._after_attempts = kwargs.get('after_attempts')
        self._wait_jitter_max = kwargs.get('wait_jitter_max')

        # Funções de parada
        stop_criterios = []
        if self._stop_max_attempt_number is not None:
            stop_criterios.append(self.parar_apos_tentativa)
        if self._stop_max_delay is not None:
            stop_criterios.append(self.parar_apos_atraso)
        self.parar_execucao = lambda num, atraso: any(c(num, atraso) for c in stop_criterios) if stop_criterios else lambda *a: False

        # Funções de espera
        wait_criterios = [lambda *a: 0] # Default: sem espera
        if self._wait_fixed is not None:
            wait_criterios.append(self.espera_fixa)
        if self._wait_random_min is not None or self._wait_random_max is not None:
            wait_criterios.append(self.espera_aleatoria)
        if self._wait_exponential_multiplier is not None or self._wait_exponential_max is not None:
            wait_criterios.append(self.espera_exponencial)
        self.calcular_espera = lambda num, atraso: max(c(num, atraso) for c in wait_criterios)


    def call(self, fn, *args, **kwargs):
        inicio_tempo = int(round(time.time() * 1000))
        contador_tentativas = 1
        while True:
            # Gancho para executar uma ação antes de cada tentativa
            if self._before_attempts:
                self._before_attempts(contador_tentativas)

            try:
                # Tenta executar a função alvo
                resultado_funcao = fn(*args, **kwargs)
                tentativa_atual = Attempt(resultado_funcao, contador_tentativas, False)
            except:
                # Captura qualquer exceção
                info_exc = sys.exc_info() # Retorna (tipo, valor, traceback)
                tentativa_atual = Attempt(info_exc, contador_tentativas, True)

            # Verifica se devemos reintentar com base no resultado ou exceção
            if not self.deve_rejeitar(tentativa_atual):
                return tentativa_atual.get(self._wrap_exception)
            
            # Gancho para executar uma ação após cada tentativa
            if self._after_attempts:
                self._after_attempts(contador_tentativas)
            
            atraso_desde_primeira_tentativa_ms = int(round(time.time() * 1000)) - inicio_tempo
            
            # Verifica se devemos parar com base nas condições de parada configuradas
            if self.parar_execucao(contador_tentativas, atraso_desde_primeira_tentativa_ms):
                if not self._wrap_exception and tentativa_atual.has_exception:
                    # Se não deve encapsular e houve exceção, re-levanta a original
                    raise tentativa_atual.get()
                else:
                    # Caso contrário, levanta RetryError
                    raise RetryError(tentativa_atual)
            else:
                # Calcula o tempo de espera antes da próxima tentativa
                tempo_espera_ms = self.calcular_espera(contador_tentativas, atraso_desde_primeira_tentativa_ms)
                if self._wait_jitter_max:
                    # Adiciona um "jitter" (aleatoriedade) ao tempo de espera
                    jitter_val = random.random() * self._wait_jitter_max
                    tempo_espera_ms = tempo_espera_ms + max(0, jitter_val)
                
                # Pausa a execução pelo tempo calculado
                time.sleep(tempo_espera_ms / 1000.0)

            contador_tentativas += 1 # Prepara para a próxima tentativa

O método call opera em um loop infinito (while True) que só é interrompido por um sucesso ou por uma condição de parada. Cada iteração representa uma tentativa. Os objetos Attempt encapsulam o resultado (seja um valor retornado ou uma exceção) e o número da tentativa, fornecendo um estado consistente para as decisões de reintento.

Lógica de Decisão: deve_rejeitar

Após cada tentativa, a biblioteca determina se deve reintentar ou não, usando o método deve_rejeitar (equivalente a should_reject). Este método combina as condições definidas por retry_on_exception e retry_on_result:

class Retrying:
    # ... (outros métodos) ...
    def deve_rejeitar(self, tentativa):
        """
        Determina se a tentativa atual deve ser rejeitada (ie, deve-se reintentar).
        Retorna True se a operação deve ser reintentada, False para sucesso.
        """
        reintentar_cond = False
        if tentativa.has_exception:
            # Se houver exceção, verifica se a exceção está nas condições de reintento
            reintentar_cond |= self._retry_on_exception(tentativa.valor[1])
        else:
            # Se não houver exceção, verifica o resultado da função
            reintentar_cond |= self._retry_on_result(tentativa.valor) 
        return reintentar_cond

A função _retry_on_exception (por padrão, reintenta em qualquer exceção) é chamada se uma exceção ocorre. Se a função não levanta uma exceção, _retry_on_result (por padrão, não reintenta em nenhum resultado) é chamada para avaliar o valor de retorno.

Estratégias de Parada

As condições de parada são configuradas no construtor da classe Retrying. Funções como parar_apos_tentativa e parar_apos_atraso são combinadas em parar_execucao. A execução é interrompida se qualquer uma dessas condições for atendida.

class Retrying:
    # ... (outros métodos) ...

    def parar_apos_tentativa(self, numero_tentativa_anterior, atraso_total_ms):
        """Para após o número de tentativas atingir o limite."""
        return numero_tentativa_anterior >= self._stop_max_attempt_number

    def parar_apos_atraso(self, numero_tentativa_anterior, atraso_total_ms):
        """Para após o atraso total desde a primeira tentativa atingir o limite."""
        return atraso_total_ms >= self._stop_max_delay

Estratégias de Espera

De forma similar às estratégias de parada, as estratégias de espera são configuradas no __init__. A função calcular_espera combina diferentes lógicas de atraso (fixo, aleatório, exponencial) e retorna o maior valor de espera calculado por elas.

class Retrying:
    # ... (outros métodos) ...

    def espera_fixa(self, numero_tentativa_anterior, atraso_total_ms):
        """Retorna um tempo de espera fixo."""
        return self._wait_fixed

    def espera_aleatoria(self, numero_tentativa_anterior, atraso_total_ms):
        """Retorna um tempo de espera aleatório."""
        return random.randint(self._wait_random_min, self._wait_random_max)

    def espera_exponencial(self, numero_tentativa_anterior, atraso_total_ms):
        """
        Calcula um tempo de espera exponencialmente crescente.
        O atraso é _wait_exponential_multiplier * 2^(número da tentativa),
        limitado por _wait_exponential_max.
        """
        multiplicador = self._wait_exponential_multiplier
        limite_maximo = self._wait_exponential_max

        # Calcula 2 elevado ao número da tentativa
        fator_exponencial = 2 ** numero_tentativa_anterior 
        
        # Calcula o resultado do atraso
        resultado_atraso = multiplicador * fator_exponencial
        
        # Aplica o limite máximo
        if resultado_atraso > limite_maximo:
            resultado_atraso = limite_maximo
        
        # Garante que o atraso não seja negativo
        if resultado_atraso < 0:
            resultado_atraso = 0
            
        return resultado_atraso