Backoff mechanism in Python

Iteration approach

import time
from botocore.exceptions import ClientError
from typing import Callable, List, Any, Dict

# backoff intervals for retry
BACKOFF_INTERVALS = [30, 60, 180]  # in seconds: 30s, 1min, 3min

# retryable DynamoDB error codes
RETRIABLE_DYNAMODB_ERRORS = [
    'InternalServerError',
    'ServiceUnavailable',
    'ProvisionedThroughputExceededException',
    'ItemCollectionSizeLimitExceededException',
    'LimitExceededException',
    'ThrottlingException',
    'TimeoutError',
    'RequestLimitExceeded',
    'ResourceNotFoundException',
    'UnrecognizedClientException',
]

def retry_with_backoff(
    func: Callable[..., Any],
    args: List[Any] = [],
    kwargs: Dict[str, Any] = {},
    retryable_errors: List[str] = RETRIABLE_DYNAMODB_ERRORS,
    backoff_intervals: List[int] = BACKOFF_INTERVALS
) -> Any:
    """Retries a function with backoff on specific errors."""
    for attempt, wait_time in enumerate(backoff_intervals, start=1):
        try:
            return func(*args, **kwargs)
        except ClientError as e:
            error_code = e.response['Error']['Code']
            if error_code in retryable_errors:
                print({
                    "message": "Retriable error encountered, retrying",
                    "attempt": attempt,
                    "wait_time": wait_time,
                    "error_code": error_code
                })
                time.sleep(wait_time)  # wait before retrying
            else:
                print({
                    "message": "Non-retriable error encountered",
                    "error_code": error_code,
                    "details": str(e)
                })
                raise e  # raise non-retriable error immediately

    # Final attempt without retry
    return func(*args, **kwargs)

Recursive Approach

import time
from botocore.exceptions import ClientError
from typing import Callable, List, Any, Dict

# backoff intervals for retry
BACKOFF_INTERVALS = [30, 60, 180]  # in seconds: 30s, 1min, 3min

# retryable DynamoDB error codes
RETRIABLE_DYNAMODB_ERRORS = [
    'InternalServerError',
    'ServiceUnavailable',
    'ProvisionedThroughputExceededException',
    'ItemCollectionSizeLimitExceededException',
    'LimitExceededException',
    'ThrottlingException',
    'TimeoutError',
    'RequestLimitExceeded',
    'ResourceNotFoundException',
    'UnrecognizedClientException',
]

def retry_with_backoff(
    func: Callable[..., Any],
    args: List[Any] = [],
    kwargs: Dict[str, Any] = {},
    retryable_errors: List[str] = RETRIABLE_DYNAMODB_ERRORS,
    backoff_intervals: List[int] = BACKOFF_INTERVALS,
    attempt: int = 1
) -> Any:
    """Retries a function with backoff on specific errors using recursion."""
    try:
        return func(*args, **kwargs)
    except ClientError as e:
        error_code = e.response['Error']['Code']
        if error_code in retryable_errors and backoff_intervals:
            wait_time = backoff_intervals[0]
            print({
                "message": "Retriable error encountered, retrying",
                "attempt": attempt,
                "wait_time": wait_time,
                "error_code": error_code
            })
            time.sleep(wait_time)  # Wait before the next retry
            # recursive call with the remaining backoff intervals
            return retry_with_backoff(
                func,
                args,
                kwargs,
                retryable_errors,
                backoff_intervals[1:],  # Remove the first interval for the next attempt
                attempt + 1
            )
        elif error_code not in retryable_errors:
            # non-retriable error, raise immediately
            print({
                "message": "Non-retriable error encountered",
                "error_code": error_code,
                "details": str(e)
            })
            raise e
        else:
            # no backoff intervals left, making the final attempt
            print({
                "message": "Final attempt after exhausting retries",
                "error_code": error_code
            })
            return func(*args, **kwargs)