""" Scanner Utilities Common utilities for AWS resource scanners including retry logic. """ import time import logging from functools import wraps from botocore.exceptions import ClientError, BotoCoreError logger = logging.getLogger(__name__) # Retry configuration MAX_RETRIES = 3 BASE_DELAY = 1.0 # seconds MAX_DELAY = 30.0 # seconds EXPONENTIAL_BASE = 2.0 def retry_with_backoff(max_retries: int = MAX_RETRIES): """ Decorator for retrying AWS API calls with exponential backoff. Requirements: - 5.5: Retry with exponential backoff up to 3 times """ def decorator(func): @wraps(func) def wrapper(*args, **kwargs): last_exception = None for attempt in range(max_retries): try: return func(*args, **kwargs) except (ClientError, BotoCoreError) as e: last_exception = e # Check if error is retryable if isinstance(e, ClientError): error_code = e.response.get('Error', {}).get('Code', '') # Non-retryable errors if error_code in ['AccessDenied', 'UnauthorizedAccess', 'InvalidParameterValue', 'ValidationError']: raise if attempt < max_retries - 1: delay = min( BASE_DELAY * (EXPONENTIAL_BASE ** attempt), MAX_DELAY ) logger.warning( f"Retry {attempt + 1}/{max_retries} for {func.__name__} " f"after {delay:.1f}s: {str(e)}" ) time.sleep(delay) else: logger.error( f"All {max_retries} retries failed for {func.__name__}: {str(e)}" ) raise last_exception return wrapper return decorator