def retry_operation()

in scripts/collect_models.py [0:0]

• 18 lines of code
• 7 McCabe index (conditional complexity)

def retry_operation(func: Callable, *args, max_retries: int = 10, initial_delay: int = 1, **kwargs) -> Any:
    """
    Retry a given operation with exponential backoff.

    Args:
        func: Callable function to execute.
        *args: Positional arguments for the function.
        max_retries: Maximum retry attempts.
        initial_delay: Initial delay between retries.
        **kwargs: Keyword arguments for the function.

    Returns:
        The function result or False if a specific RuntimeError occurs.

    Raises:
        Exception if max retries are exceeded or on non-retryable errors.
    """
    delay = initial_delay
    for attempt in range(max_retries):
        try:
            return func(*args, **kwargs)
        except RuntimeError:
            return False
        except (RequestException, HfHubHTTPError) as e:
            status = getattr(e, "response", None)
            if status is not None and (status.status_code == 429 or 500 <= status.status_code < 600):
                logging.warning(
                    f"Attempt {attempt + 1} failed with status {status.status_code}. Retrying after {delay} seconds..."
                )
                time.sleep(delay)
                delay *= 2
            else:
                raise
    raise Exception("Max retries exceeded while trying to execute operation.")