python retry装饰器使用

第一个例子

import time
from functools import wraps
from typing import Callable, Any
from time import sleep


def retry(retries: int = 3, delay: float = 1) -> Callable:
    """
    Attempt to call a function, if it fails, try again with a specified delay.

    :param retries: The max amount of retries you want for the function call
    :param delay: The delay (in seconds) between each function retry
    :return:
    """

    # Don't let the user use this decorator if they are high
    if retries < 1 or delay <= 0:
        raise ValueError('Are you high, mate?')

    def decorator(func: Callable) -> Callable:
        @wraps(func)
        def wrapper(*args, **kwargs) -> Any:
            for i in range(1, retries + 1):  # 1 to retries + 1 since upper bound is exclusive

                try:
                    print(f'Running ({i}): {func.__name__}()')
                    return func(*args, **kwargs)
                except Exception as e:
                    # Break out of the loop if the max amount of retries is exceeded
                    if i == retries:
                        print(f'Error: {repr(e)}.')
                        print(f'"{func.__name__}()" failed after {retries} retries.')
                        break
                    else:
                        print(f'Error: {repr(e)} -> Retrying...')
                        sleep(delay)  # Add a delay before running the next iteration

        return wrapper

    return decorator


@retry(retries=3, delay=1)
def connect() -> None:
    time.sleep(1)
    raise Exception('Could not connect to internet...')


def main() -> None:
    connect()


if __name__ == '__main__':
    main()

代码解析

1. 导入

该代码使用了以下库：

time: 提供时间相关操作函数，用于延迟和模拟操作。
functools: 包含高阶函数相关工具，其中包括用于保留函数元数据的 wraps 装饰器。
typing: 用于类型注释，以增强代码的可读性和可维护性。

2. `retry` 装饰器

用途

该装饰器用于对装饰后的函数进行指定次数的重试，并在重试之间设置延迟时间。

参数

retries (int, 默认值=3): 最大重试次数。
delay (float, 默认值=1): 重试之间的延迟时间（秒）。

功能

检查 retries 或 delay 是否无效。
使用 wraps 保留装饰函数的元数据。
使用循环来重试函数：
- 打印当前尝试次数的提示信息。
- 调用装饰函数。
- 如果成功，返回函数的返回值。
- 如果发生异常：
  - 打印错误信息。
  - 如果重试次数已用完，报告失败。
  - 否则，延迟 delay 秒并重试。

3. `connect` 函数

用途

模拟一个连接尝试，最终以异常失败。

装饰器

使用 @retry(retries=3, delay=1) 装饰器启用重试。

行为

延迟 1 秒以模拟连接尝试。
抛出异常以指示失败。

4. `main` 函数

用途

程序的入口点。

行为

调用 connect 函数，触发重试过程。

5. `if name == 'main':`

用途

确保 main 函数只有在直接运行脚本时才会执行，而不会在将其作为模块导入时执行。

操作

调用 main 函数来启动执行。

输出

该代码将产生如下输出：

Running (1): connect()
Error: Could not connect to internet... -> Retrying...
Running (2): connect()
Error: Could not connect to internet... -> Retrying...
Running (3): connect()
Error: Could not connect to internet...
"connect()" failed after 3 retries.

第二个例子

import random
from time import sleep

def retry(retries=3, delay=1):
    def decorator(func):
        def wrapper():
            for i in range(retries):
                try:
                    result = func()
                    return result
                except Exception as e:
                    print(f"Attempt {i+1} failed: {e}")
                    sleep(delay)  # Adding a delay before retrying
            return None  # Return None if all retries fail
        return wrapper
    return decorator

@retry()
def unpredictable_function():
    if random.random() < 0.5:
        raise ValueError("Random error occurred")
    return "Success!"

result = unpredictable_function()
print(result)