Loading Academy Hub...
Build industrial-grade Python decorators for automatic exponential backoff with jitter on API rate limits, and contextual JSON structured logging for Datadog and ELK.
[!NOTE] In production pipelines, 80% of pipeline downtime stems from transient network blips, 429 Too Many Requests, and non-actionable plain-text logs. Implementing clean function decorators allows decoupling infrastructure resilience from business transformation logic.
Blind retries at fixed intervals cause the Thundering Herd Problem, where hundreds of concurrent workers hammer an overloaded database in synchrony. Full jitter disperses the retry spikes:
functools.wraps(func): Never write a custom decorator without @wraps(func)! Omitting it erases __name__, __doc__, and breaks pytest, Sphinx docs, and FastAPI dependency introspection.400 Bad Request or 403 Forbidden (permanent auth/payload bugs!). Retry only transient faults (429 Too Many Requests, 502/503/504, timeouts) with exponential backoff + random jitter (tenacity) so 50 failed workers do not DDOS a recovering database simultaneously (Thundering Herd).trace_id, run_id, batch_date): Plain string logs cannot be queried efficiently in CloudWatch/Datadog; emit single-line JSON dicts cleanly.Review the production code, company interview Q&As, incident traces, and comprehension quiz inside this lesson.
import time
import random
import logging
import json
import functools
from typing import Callable, Any
logger = logging.getLogger("pipeline")
logger.setLevel(logging.INFO)
def retry_with_backoff(max_retries: int = 3, base_delay: float = 1.0, max_delay: float = 30.0):
def decorator(func: Callable) -> Callable:
@functools.wraps(func)
def wrapper(*args, **kwargs) -> Any:
attempt = 0
while attempt < max_retries:
try:
return func(*args, **kwargs)
except Exception as ex:
attempt += 1
if attempt >= max_retries:
logger.error(json.dumps({
"event": "retry_exhausted",
"function": func.__name__,
"attempts": attempt,
"error": str(ex)
}))
raise
sleep_time = min(max_delay, base_delay * (2 ** (attempt - 1)))
jitter = random.uniform(0, sleep_time)
logger.warning(json.dumps({
"event": "retry_attempt",
"function": func.__name__,
"attempt": attempt,
"sleep_s": round(jitter, 2),
"reason": str(ex)
}))
time.sleep(jitter)
return wrapper
return decoratorTypeError: missing 1 required positional argument