Created
October 3, 2025 10:10
-
-
Save musoftware/1a5d173a773282d45eab7481ded309ad to your computer and use it in GitHub Desktop.
Generic Retry Logic as a Reusable Class
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| import time | |
| import random | |
| import logging | |
| from functools import wraps | |
| # Set up basic logging | |
| logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') | |
| class Retry: | |
| """ | |
| A generic decorator class to retry a function with exponential backoff. | |
| This class is initialized with retry parameters and can be used to | |
| decorate any function that might fail due to transient errors. | |
| """ | |
| def __init__(self, retries=3, base_delay=1.0, backoff_factor=2.0, jitter=0.1): | |
| """ | |
| Initializes the decorator with retry configuration. | |
| """ | |
| self.retries = retries | |
| self.base_delay = base_delay | |
| self.backoff_factor = backoff_factor | |
| self.jitter = jitter | |
| def __call__(self, func): | |
| """ | |
| This method is called when the decorator is applied to a function. | |
| It returns the wrapper that contains the retry logic. | |
| """ | |
| @wraps(func) | |
| def wrapper(*args, **kwargs): | |
| delay = self.base_delay | |
| for attempt in range(self.retries): | |
| try: | |
| return func(*args, **kwargs) | |
| except Exception as e: | |
| if attempt == self.retries - 1: | |
| logging.error(f"Function '{func.__name__}' failed after {self.retries} retries.") | |
| raise | |
| sleep_time = delay + random.uniform(0, self.jitter) | |
| logging.warning( | |
| f"Attempt {attempt + 1}/{self.retries} failed for '{func.__name__}': {e}. " | |
| f"Retrying in {sleep_time:.2f} seconds..." | |
| ) | |
| time.sleep(sleep_time) | |
| delay *= self.backoff_factor # Increase delay for the next attempt | |
| return wrapper | |
| # --- Example Usage --- | |
| # The usage looks identical, but it's now powered by a class instance. | |
| @Retry(retries=5, base_delay=1, backoff_factor=2, jitter=0.5) | |
| def get_database_connection(): | |
| """Attempts to connect to a database that might be busy.""" | |
| print("Attempting to connect to the database...") | |
| if random.random() < 0.8: # 80% chance of failure | |
| raise TimeoutError("Database connection timed out") | |
| print("Database connection successful!") | |
| return {"status": "connected"} | |
| # Run the function | |
| if __name__ == "__main__": | |
| try: | |
| connection = get_database_connection() | |
| print(f"Final result: {connection}") | |
| except TimeoutError as e: | |
| print(f"Failed to connect to database: {e}") |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment