Skip to Content
Course content

155: Practice Exercise: Building a Custom Retry Decorator for Flaky Operations

Click on the "Edit" button in the top corner of the screen to edit your slide content.

Look, we've all been there. You're calling an external API, or maybe hitting a database that's under heavy load, and every once in a while, the request just... fails. A ConnectionError pops up, your script crashes, and you're left wondering why your code worked perfectly in staging but is dying in production. Instead of wrapping every single network call in a messy while loop, we're going to build a reusable retry decorator. It's a much cleaner way to handle "flaky" operations.

Simulating the Flakiness

Before we build the solution, we need a problem to solve. I'll write a quick function that simulates a flaky API call. I'm using the random module here to make it fail about 70% of the time. This gives us something concrete to test against.

import random

def fetch_api_data():
    print("Attempting to fetch data...")
    if random.random() < 0.7:
        print("--- Network glitch! ---")
        raise ConnectionError("Server didn't respond")
    print("Success! Data received.")
    return {"status": "200", "data": "Hello World"}

The First Naive Attempt

My first instinct is to write a simple decorator that catches the exception and tries one more time. I'll keep it basic: if it fails, try once more, and if it fails again, let the exception bubble up. This is the skeleton of our logic.

import functools

def retry(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except ConnectionError:
            print("Retrying once...")
            return func(*args, **kwargs)
    return wrapper

@retry
def fetch_api_data():
    # ... (same as above)

Where I Messed Up the Arguments

The problem with the version above is that it's too rigid. Some APIs need three retries; some only need one. I wanted to be able to do something like @retry(times=3). I tried to just add a times argument to my existing retry function, like this:

# DON'T DO THIS - This is my mistake!
def retry(func, times=3):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        # ... logic to loop 'times' ...
        pass
    return wrapper

When I tried to run this as @retry(times=3), Python threw a TypeError. Why? Because when you put parentheses after a decorator, Python expects the decorator to be a factory—a function that returns the actual decorator. I was trying to use the decorator itself as the factory. I forgot that adding parameters to a decorator requires an extra layer of nesting.

Getting the Nesting Right

To fix this, I need three levels: the factory (which takes the configuration), the decorator (which takes the function), and the wrapper (which handles the execution). It feels like a lot of boilerplate, but it's the only way to pass arguments into the decorator logic.

import functools
import time

def retry(times=3, delay=1):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            attempts = 0
            while attempts < times:
                try:
                    return func(*args, **kwargs)
                except ConnectionError as e:
                    attempts += 1
                    if attempts >= times:
                        print("All retries exhausted.")
                        raise e
                    print(f"Attempt {attempts} failed. Retrying in {delay}s...")
                    time.sleep(delay)
        return wrapper
    return decorator

Adding a Little Breathing Room

I added time.sleep(delay) in the corrected version above. In the real world, hammering a failing server as fast as your CPU can loop usually makes the problem worse—it's basically a self-inflicted DDoS attack. By adding a small delay, we give the remote service a moment to recover.

Now, we can apply this to our flaky function with specific settings:

@retry(times=5, delay=2)
def fetch_api_data():
    print("Attempting to fetch data...")
    if random.random() < 0.7:
        print("--- Network glitch! ---")
        raise ConnectionError("Server didn't respond")
    print("Success! Data received.")
    return {"status": "200", "data": "Hello World"}

# Now we call it, and it'll try up to 5 times before finally giving up.
print(fetch_api_data())



📋 Practical Task

Exercise: Implementing Exponential Backoff for the Retry Decorator

In the previous lesson, we used a constant delay between retries. However, in professional production environments, it's standard practice to use Exponential Backoff. Instead of waiting 1 second every time, the wait time doubles with each failure (e.g., 1s, 2s, 4s, 8s). This prevents a "thundering herd" problem where many clients retry at the exact same interval.

Your Task: Modify the retry decorator from the lesson so that the delay increases exponentially after each failure.

  • The delay parameter should represent the initial wait time.
  • Each subsequent retry should multiply the current delay by 2.
  • Ensure the decorator still accepts times and delay as arguments.
  • Print the current wait time to the console so you can verify the doubling effect (e.g., "Retrying in 1s...", "Retrying in 2s...", "Retrying in 4s...").
Rating
0 0

There are no comments for now.

to be the first to leave a comment.