SolveConPython

Python Reto #27 — Retry con backoff exponencial (sin librerías)

Nivel: Intermedio+
Tema: Resiliencia, reintentos, backoff exponencial, jitter opcional, diseño de APIs, tests deterministas
Objetivo: Implementar una función de reintentos robusta para operaciones que pueden fallar temporalmente (red, IO, APIs), sin depender de librerías externas.

Enunciado

Crea una función llamada:

retry(func, intentos=3, delay_inicial=0.1, factor=2.0, excepciones=(Exception,), sleep_fn=None)

Que:

  1. Reciba:
    • func: callable sin argumentos (lo llamaremos como func())
    • intentos: número máximo de intentos (int)
    • delay_inicial: tiempo inicial de espera en segundos (float/int)
    • factor: multiplicador de backoff (float/int)
    • excepciones: tupla de clases de excepción a capturar
    • sleep_fn: función para dormir (por defecto time.sleep) para poder testear sin esperar
  2. Validación:
    • func debe ser callable o TypeError
    • intentos debe ser int >= 1 o TypeError/ValueError
    • delay_inicial debe ser numérico >= 0 o TypeError/ValueError
    • factor debe ser numérico > 0 o TypeError/ValueError
    • excepciones debe ser una tupla de excepciones o TypeError
    • sleep_fn si se pasa debe ser callable o TypeError
  3. Comportamiento:
    • Llama a func() hasta intentos veces.
    • Si func() funciona, devuelve su resultado inmediatamente.
    • Si func() lanza una excepción incluida en excepciones:
      • si aún quedan intentos, espera delay segundos y reintenta,
      • multiplica el delay por factor para el siguiente intento.
    • Si se agotan los intentos, debe relanzar la última excepción capturada.
    • Si func() lanza una excepción que no está en excepciones, debe propagarse sin reintentos.

Ejemplos

Python
# reintenta 3 veces: delays 0.1s, 0.2s (y luego falla si sigue fallando)
retry(mi_func, intentos=3, delay_inicial=0.1, factor=2.0)
# capturar solo ValueError
retry(mi_func, excepciones=(ValueError,))

Pistas

  1. Usa un bucle for o while contando intentos.
  2. Guarda el delay actual en una variable y actualízalo multiplicando por factor.
  3. Inyectar sleep_fn permite tests sin pausas reales.

Solución explicada (paso a paso)

  1. Validar parámetros.
  2. Inicializar delay = delay_inicial.
  3. Repetir:
    • intentar ejecutar func()
    • si falla con excepción capturable:
      • si es el último intento → relanzar
      • si no → dormir delay y actualizar delay *= factor
  4. Devolver resultado si éxito.

Python
import time
def retry(
    func,
    intentos: int = 3,
    delay_inicial: float = 0.1,
    factor: float = 2.0,
    excepciones: tuple = (Exception,),
    sleep_fn=None,
):
    """
    Ejecuta func() con reintentos y backoff exponencial.
    - Captura solo excepciones indicadas en 'excepciones'
    - Espera delay entre intentos, multiplicándolo por 'factor'
    - Si se agotan intentos, relanza la última excepción capturada
    """
    if not callable(func):
        raise TypeError("El parámetro 'func' debe ser callable.")
    if intentos is None or not isinstance(intentos, int):
        raise TypeError("El parámetro 'intentos' debe ser un entero (int).")
    if intentos < 1:
        raise ValueError("'intentos' debe ser >= 1.")
    if delay_inicial is None or not isinstance(delay_inicial, (int, float)):
        raise TypeError("El parámetro 'delay_inicial' debe ser numérico (int o float).")
    if delay_inicial < 0:
        raise ValueError("'delay_inicial' debe ser >= 0.")
    if factor is None or not isinstance(factor, (int, float)):
        raise TypeError("El parámetro 'factor' debe ser numérico (int o float).")
    if factor <= 0:
        raise ValueError("'factor' debe ser > 0.")
    if not isinstance(excepciones, tuple) or not excepciones:
        raise TypeError("El parámetro 'excepciones' debe ser una tupla no vacía de excepciones.")
    # Validar que sean clases de excepción
    for exc in excepciones:
        if not isinstance(exc, type) or not issubclass(exc, BaseException):
            raise TypeError("'excepciones' debe contener solo clases de excepción.")
    if sleep_fn is None:
        sleep_fn = time.sleep
    elif not callable(sleep_fn):
        raise TypeError("El parámetro 'sleep_fn' debe ser callable si se proporciona.")
    delay = float(delay_inicial)
    ultimo_error = None
    for intento in range(1, intentos + 1):
        try:
            return func()
        except excepciones as e:
            ultimo_error = e
            if intento == intentos:
                raise  # relanza la última excepción capturada
            if delay > 0:
                sleep_fn(delay)
            delay *= float(factor)
    # No debería llegar aquí, pero por seguridad:
    if ultimo_error:
        raise ultimo_error

Python
import pytest
from reto_27_retry_backoff import retry
class SleepFake:
    def __init__(self):
        self.calls = []
    def sleep(self, seconds):
        self.calls.append(seconds)
def test_retry_exito_despues_de_fallos():
    sleep = SleepFake()
    estado = {"n": 0}
    def f():
        estado["n"] += 1
        if estado["n"] < 3:
            raise ValueError("fallo temporal")
        return "ok"
    res = retry(f, intentos=5, delay_inicial=0.1, factor=2.0, excepciones=(ValueError,), sleep_fn=sleep.sleep)
    assert res == "ok"
    assert estado["n"] == 3
    assert sleep.calls == [0.1, 0.2]  # esperó entre intentos fallidos
def test_retry_relanzar_ultima_excepcion():
    sleep = SleepFake()
    def f():
        raise ValueError("siempre falla")
    with pytest.raises(ValueError):
        retry(f, intentos=3, delay_inicial=0.1, factor=2.0, excepciones=(ValueError,), sleep_fn=sleep.sleep)
    assert sleep.calls == [0.1, 0.2]  # no duerme después del último intento
def test_excepcion_no_capturada_no_reintenta():
    sleep = SleepFake()
    estado = {"n": 0}
    def f():
        estado["n"] += 1
        raise KeyError("no capturada")
    with pytest.raises(KeyError):
        retry(f, intentos=5, delay_inicial=0.1, factor=2.0, excepciones=(ValueError,), sleep_fn=sleep.sleep)
    assert estado["n"] == 1
    assert sleep.calls == []
def test_validaciones():
    with pytest.raises(TypeError):
        retry(123)
    with pytest.raises(ValueError):
        retry(lambda: 1, intentos=0)
    with pytest.raises(ValueError):
        retry(lambda: 1, delay_inicial=-1)
    with pytest.raises(ValueError):
        retry(lambda: 1, factor=0)
    with pytest.raises(TypeError):
        retry(lambda: 1, excepciones="no_tuple")
    with pytest.raises(TypeError):
        retry(lambda: 1, sleep_fn="no_callable")

Ejecuta:

  • pytest -q

Variantes para subir de nivel (opcional)

  1. Jitter: añadir aleatoriedad al delay para evitar “thundering herd”
  2. Callback on_retry: función que reciba (intento, excepción, delay)
  3. Soportar func(*args, **kwargs) en lugar de solo func()
  4. Límite máximo de delay (max_delay)
  5. Retry condicional según el mensaje/código de error

Lo que aprendiste

  • Diferencia entre fallos temporales vs. fallos definitivos
  • Cómo controlar reintentos sin ocultar errores no previstos
  • Backoff exponencial y tests deterministas con sleep_fn
  • Diseño de una utilidad reusable “tipo producción”

Accede al código completo y a los tests en GitHub para ejecutar y modificar la solución localmente.

Siguiente reto recomendado: Reto #28 — Circuit Breaker simple (cuando dejar de reintentar y “abrir” el circuito).