Python模块 adorner 的使用示例

模块介绍#

adorner 是一个现代轻量级的 Python 装饰器辅助模块。

目前该模块仅实现了 4 个类，对应着 4 个功能：制造装饰器、执行计时、函数缓存、捕获重试。

仓库地址：https://github.com/gupingan/adorner

安装#

该模块可在上方仓库中的 Releases 页面下载 tar.gz 文件后离线安装，也可以通过包管理工具进行下载安装：

pip install adorner

也可以尝试下方这个命令：

pip install adorner -i http://mirrors.cloud.tencent.com/pypi/simple/

或者更换为： https://mirrors.cloud.tencent.com/pypi/simple/

---

类 Decorator#

Decorator 类用于标记装饰器函数，使装饰器的构造更简单。它允许你定义一个装饰器并将其应用到函数上，简化了装饰器的创建和使用过程。

源码注释#

class Decorator(object):
    def __init__(self, decorator=None):
        self.decorator = decorator or (lambda s: s.execute())  # 是传入的装饰器函数，如果没有传入，则默认使用一个简单的 lambda 函数，该函数调用 self.execute()。
        self.function = None  # 用于存储被装饰的函数。
        # args 和 .kwargs 分别用于存储传递给被装饰函数的位置参数和关键字参数。
        self.args = tuple()
        self.kwargs = dict()

    def __call__(self, function):
        """
        这里是关键部分
        __call__ 可使得类的实例可以像函数一样被调用。
        接收一个函数 function 作为参数，并返回一个 wrapper 函数。
        wrapper 函数内部将被装饰的函数及其参数存储在 self.function、self.args 和 self.kwargs 中，然后调用 self.decorator(self)。
        functools.wraps(function) 用于保持被装饰函数的元数据（如函数名和文档字符串）
        """
        @functools.wraps(function)
        def wrapper(*args, **kwargs):
            self.function = function
            self.args = args
            self.kwargs = kwargs
            return self.decorator(self)

        return wrapper

    def __repr__(self):
        """
        返回对象的字符串表示形式，便于调试和查看对象信息。根据是否有被装饰的函数来返回不同的字符串。
        """
        decorator_name = self.decorator_name.lstrip('<').rstrip('>')
        if self.function:
            return f'<{self.__class__.__name__}: {decorator_name} To {self.function.__name__}>'
        return f'<{self.__class__.__name__}: {decorator_name}>'

    def execute(self, *args, **kwargs):
        """
        用于执行被装饰的函数。会使用传入的参数（如果有）或存储的参数来调用 _execute_sync 方法，该方法应该是为了以后适配更复杂的异步装饰器所提前编写好的
        """
        final_args = args if args else self.args
        final_kwargs = kwargs if kwargs else self.kwargs
        return self._execute_sync(final_args, final_kwargs)

    def _execute_sync(self, args, kwargs):
        """
        同步地执行被装饰的函数，并返回其结果
        """
        return self.function(*args, **kwargs)

    @property
    def function_name(self):
        """返回被装饰函数的名称"""
        if self.function:
            return self.function.__name__
        return ''

    @property
    def function_doc(self):
        """返回被装饰函数的文档字符串"""
        if self.function:
            return self.function.__doc__ or ''
        return ''

    @property
    def decorator_name(self):
        """返回装饰器的名称"""
        if self.decorator:
            return self.decorator.__name__
        return ''

    @property
    def decorator_doc(self):
        """返回装饰器的文档字符串"""
        if self.decorator:
            return self.decorator.__doc__ or ''
        return ''

示例用法#

import time
from adorner import Decorator


@Decorator
def exception_decorator(self: Decorator):
    """
    捕获异常日志的装饰器
    :param self: 装饰器 Decorator 实例
    :return: 被修饰函数的执行结果
    """
    print(self.function_doc)  # 打印被装饰函数的文档
    print(self.decorator_doc)  # 打印装饰器的文档
    print(self.function_name)  # 打印被装饰函数的名称
    print(self.decorator_name)  # 打印装饰器的名称
    print(self.args)  # 打印被装饰函数的传入的位置参数 （默认形参值不包含）
    print(self.kwargs)  # 打印被装饰函数的传入的关键字参数  （默认形参值不包含）
    try:
        result = self.execute()  # 打印 1
        # 执行被装饰函数，不传入任何参数时，表示使用默认的参数 self.args、self.kwargs
        # 可覆盖传入参数
        self.execute(value=2)  # 打印 2
        self.execute(3)  # 打印3 并抛出异常
        return result
    except Exception as e:
        print(f"捕获异常: {e}")
        raise


@exception_decorator
def risky_function(value=1):
    print(value)
    if value == 3:
        raise ValueError("出错了")


try:
    risky_function()
except ValueError:
    pass  # 捕获异常: 出错了

上述示例执行后，终端应该会输出：

    捕获异常日志的装饰器
    :param self: 装饰器 Decorator 实例
    :return: 被修饰函数的执行结果
    
risky_function
exception_decorator
()
{}
1
2
3
捕获异常: 出错了

---

类 Timer#

Timer 类是 Decorator 类的一个子类，用于测量被装饰函数的执行时间。它继承了 Decorator 类的所有功能，并在执行函数时记录开始和结束的时间，以计算函数的执行时长，该类属于 Decorator 类的扩展使用。

源码注释#

class Timer(Decorator):
    def __init__(self, decorator=None):
        super().__init__(decorator)  # 调用父类 Decorator 的构造函数，初始化装饰器函数。
        self.time = 0  # 用于存储被装饰函数的执行时间。

    def execute(self, *args, **kwargs):
        """
        执行被装饰的函数，并记录其执行时间。
        使用 time.perf_counter() 记录开始和结束的时间，计算函数执行时长，并存储在 self.time 中。
        """
        _start = time.perf_counter()  # 记录开始时间。
        result = super().execute(*args, **kwargs)  # 调用父类的 execute 方法执行被装饰的函数。
        _end = time.perf_counter()  # 记录结束时间。
        self.time = _end - _start  # 计算并存储执行时间。
        return result  # 返回被装饰函数的结果。

示例用法#

下面是如何使用 Timer 类来装饰一个函数，并测量其执行时间的示例：

import time
from adorner import Timer

timer = Timer()  # 可装饰多个函数，不过不太推荐（多个函数先后执行会覆盖掉计时器的元数据）


@timer
def my_function(a, b):
    """一个简单的函数，用于演示 Timer 装饰器的使用。"""
    time.sleep(1)  # 模拟一个耗时操作。
    return a + b


result = my_function(1, 2)
print(f'Execution result: {result}')
print(f"Execution time: {timer.time} seconds")

输出将类似于：

Execution result: 3
Execution time: 1.0067455 seconds

---

类 Cacher#

Cacher 类是一个装饰器类，用于管理和缓存函数对象及其相关数据，函数不仅仅是函数，本身也是轻量级的缓存器。

源码注释#

class Cacher:
    hash = dict()  # 用于存储每个被装饰函数的 Cacher 实例。

    def __new__(cls, function):
        """
        确保每个被装饰的函数只有一个 Cacher 实例。
        如果该函数已经有一个 Cacher 实例，则返回该实例；
        否则，创建一个新的实例，并将其存储在 hash 中。
        """
        if function in cls.hash:
            instance = cls.hash[function]
        else:
            instance = object.__new__(cls)
            instance.function = function  # 设置缓存实例对应的函数
            instance.data = dict()  # 缓存存储的结构是字典
            setattr(instance, '__name__', f'{cls.__name__}-{function.__name__}')
            cls.hash[function] = instance

        return instance

    def __call__(self, *args, **kwargs):
        """
        使 Cacher 实例可以像函数一样被调用。
        调用被装饰的函数，并返回其结果。
        """
        return self.function(*args, **kwargs)

    def __repr__(self):
        """
        返回对象的字符串表示形式，便于调试和查看对象信息。
        """
        return f'<{self.__class__.__name__}: {self.function.__name__}>'

    def __iter__(self):
        """
        使 Cacher 实例可迭代，迭代缓存数据。
        """
        return iter(self.data)

    def __contains__(self, item):
        """
        判断缓存数据中是否包含指定的键。
        """
        return item in self.data

    def __add__(self, other):
        """
        支持使用 + 运算符合并缓存数据。
        """
        if isinstance(other, self.__class__):
            self.data.update(other.data)
            return self
        if isinstance(other, dict):
            self.data.update(other)
            return self
        if isinstance(other, (tuple, list)):
            self.data.update(dict(other))
            return self
        raise TypeError(f'unsupported operand type(s) for +: \'{type(self)}\' and \'{type(other)}\'')

    def __sub__(self, other):
        """
        支持使用 - 运算符从缓存数据中删除指定的键。
        """
        if isinstance(other, self.__class__):
            for key in other.data:
                self.data.pop(key, None)
            return self
        if isinstance(other, dict):
            for key in other:
                self.data.pop(key, None)
            return self
        if isinstance(other, (tuple, list)):
            self.pops(*other)
            return self
        raise TypeError(f'unsupported operand type(s) for -: \'{type(self)}\' and \'{type(other)}\'')
    
    def items(self):
        return self.data.items()
    
    def set(self, key, value, safe=False):
        """
        设置缓存数据。
        如果 safe 为 True，则只有在 key 不存在的情况下才设置值。
        """
        if not safe:
            self.data[key] = value
        elif key not in self.data:
            self.data[key] = value

        return self.data[key]

    def sets(self, **data_dict):
        """
        批量设置缓存数据。
        """
        self.data.update(data_dict)

    def get(self, key, default_value=None):
        """
        获取缓存数据。
        如果 key 不存在，则返回 default_value。
        """
        return self.data.get(key, default_value)

    
    @staticmethod
    def _apply_filter(values, filter_function, filter_safe, filter_errors):
        """应用筛选函数"""
        def safe_filter(value):
            try:
                return filter_function(value)
            except filter_errors:
                return False

        filter_func = safe_filter if filter_safe else filter_function
        return {key: value for key, value in values.items() if filter_func(value)}

    @staticmethod
    def _apply_map(values, map_function, map_safe, map_errors):
        """应用遍历处理的函数"""
        def safe_map(value_):
            try:
                return True, map_function(value_)
            except map_errors:
                return False, None

        if map_safe:
            new_values = {}
            for key, value in values.items():
                success, mapped_value = safe_map(value)
                if success:
                    new_values[key] = mapped_value
            return new_values
        else:
            return {key: map_function(value) for key, value in values.items()}

    
    def gets(self, *keys, default_value=None, filter_function=None, map_function=None):
        """
        批量获取缓存数据。
        支持通过 filter_function 过滤值，通过 map_function 处理值。
        """
        values = {key: self.data.get(key, default_value) for key in keys}

        if filter_function:
            filter_errors = filter_errors or (TypeError, ValueError, KeyError, IndexError)
            values = self._apply_filter(values, filter_function, filter_safe, filter_errors)

        if map_function:
            map_errors = map_errors or (TypeError, ValueError, KeyError, IndexError)
            values = self._apply_map(values, map_function, map_safe, map_errors)

        return values

    def pop(self, key, default_value=None):
        """
        删除并返回缓存数据中的指定键。
        如果键不存在，则返回 default_value。
        """
        return self.data.pop(key, default_value)

    def pops(self, *keys, default_value=None):
        """
        批量删除并返回缓存数据中的指定键。
        如果键不存在，则返回 default_value。
        """
        return [self.data.pop(key, default_value) for key in keys]

使用案例#

下面是如何使用 Cacher 类来装饰函数，并操作缓存数据的示例：

from adorner import Cacher


@Cacher
def example1(x):
    """计算乘积"""
    return x * x


@Cacher
def example2(x):
    """计算和"""
    return x + x


print(example1)  # 打印：
# 正常调用
print(example1(4))  # 打印：16
# 打印函数的文档字符串
print(example1.function_doc)

# 缓存设置数据
example1.set('a', 1)
example1.set('b', 2)
example1.set('c', 3)

# example2.set('a', True)
# example2.set('b', False)
# 和上述一致
example2.sets(a=True, b=False, d='数据 d')

# 获取缓存数据
print(example1.get('a'))
print(example1.get('d', '数据不存在'))
# 检查 d 是否在缓存器 example1 中
print('d' in example1)

# 缓存数据合并
new_cacher = example1 + example2
print(new_cacher.data)  # 缓存器的所有数据
# 打印：{'a': True, 'b': False, 'c': 3, 'd': '数据 d'}

print(list(new_cacher))  # 将缓存器转为列表，可呈现存储的键

new_cacher += {'e': '合并的数据 e'}
# 迭代打印
for k, v in new_cacher.items():
    print(k, v)

# 批量获取数据
print(new_cacher.gets('a', 'b', 'z', default_value='没有这个数据'))
print(new_cacher.gets('a', 'b', 'c', filter_function=lambda x: x > 1))
# 如果比较类型不一致，可能会发生错误，比如下面这个例子：
# print(new_cacher.gets('a', 'b', 'c', 'd', filter_function=lambda x: x > 1))
# 解决方式：你可以自行捕捉，但是那样会很繁琐，推荐使用 filter_safe 参数
print(new_cacher.gets('a', 'b', 'c', 'd', filter_function=lambda x: x > 1, filter_safe=True))
# 如果启用了 filter_safe 参数还无法正常捕捉，请使用 filter_errors 指定异常，默认是 (TypeError, ValueError, KeyError, IndexError)
print(new_cacher.gets('a', 'b', 'c', 'd', filter_function=lambda x: x(),
                      filter_safe=True, filter_errors=(TypeError, ValueError, KeyError, IndexError)))

# 除了上述的 filter_function 参数，另外还有 map_function，同理也有 map_safe 以及 map_errors 参数
print(new_cacher.gets('a', 'b', 'c', map_function=lambda x: x > 1))
print(new_cacher.gets('a', 'b', 'c', 'd', map_function=lambda x: x > 1, map_safe=True))
print(new_cacher.gets('a', 'b', 'c', 'd', map_function=lambda x: x > 1, map_safe=True, map_errors=(TypeError,)))

# xxx_safe 参数的功能是当传入的函数执行发生异常时对应的一个处理，当出现异常时，该值对应的键值都不应存在于结果中
# 优先级别：正常获取值 -> filter筛选 -> map遍历处理 -> 返回结果

# 弹出某个值
print(new_cacher.pop('c'))
print(new_cacher.pop('c', default_value=None))  # 上面弹出了，这里尝试弹出一个不存在的，将返回 default_value（默认None）
print(new_cacher.pop('c') == new_cacher.pop('c', default_value=None))
print(new_cacher.data)  # {'a': True, 'b': False, 'd': '数据 d', 'e': '合并的数据 e'}

# 批量弹出
print(new_cacher.pops('b', 'c', default_value='不存在'))
print(new_cacher.data)  # {'a': True, 'd': '数据 d', 'e': '合并的数据 e'}

# 减法删除
sub = new_cacher - []  # 支持减去 字典 {'a', 任意值} 以及元组 ('a',)
print(sub.data)  # {'d': '数据 d', 'e': '合并的数据 e'}
print(new_cacher.data)  # {'d': '数据 d', 'e': '合并的数据 e'}

---

类 Retryer#

Retryer 类是一个装饰器类，用于在指定异常发生时重试被装饰函数的执行。它允许设置最大重试次数、重试间隔时间以及需要捕获的异常类型。该类为函数添加了自动重试机制，适用于网络请求、文件操作等可能会临时失败的操作。

源码注释#

from typing import Union, List, Type
import time

class Retryer:
    def __init__(self, max_retry: Union[int] = 3, delay: Union[int] = 0, catches: List[Type[Exception]] = None):
        """
        初始化 Retryer 实例。
        
        :param max_retry: 最大重试次数，默认为 3。
        :param delay: 每次重试之间的延迟时间（秒），默认为 0。
        :param catches: 需要捕获的异常类型列表，默认为空列表。
        """
        self.max_retry = max_retry
        self.delay = delay
        self.catches = catches or []
        self.exceptions = []
        self.count = 0

    def __call__(self, function):
        """使 Retryer 实例可作为装饰器使用。"""
        return Decorator(self.run)(function)

    def run(self, decorator: Decorator):
        """执行重试逻辑。"""
        _catch_exceptions = tuple(self.catches) if self.catches else Exception
        self.exceptions.clear()
        i = 0
        while i <= self.max_retry:
            self.count = i
            try:
                result = decorator.execute()
            except _catch_exceptions as e:
                self.exceptions.append(e)
                i += 1
                if i <= self.max_retry:
                    time.sleep(self.delay)
                continue
            else:
                return result
        raise self.exceptions[-1]

示例用法#

下面是如何使用 Retryer 类来装饰一个函数，并在指定异常发生时重试的示例：

import random
from adorner import Retryer


# 创建 Retryer 实例，设置捕获的异常类型为 KeyError，当被装饰的函数中出现该错误时将进行重试
retryer = Retryer(catches=[KeyError])


@retryer
def unreliable_function():
    """一个可能会抛出异常的函数，用于演示 Retryer 装饰器的使用"""
    option = random.randint(0, 2)
    if option == 0:
        raise KeyError('Random KeyError')
    elif option == 1:
        raise ValueError('Random ValueError')
    else:
        return "Success"


try:
    result = unreliable_function()
    print(result)
except Exception as e:
    print(f"Function failed after retries: {e}")

# 打印重试次数和捕获的异常
print(f"Retry count: {retryer.count}")
print(f"Exceptions: {retryer.exceptions}")

输出将类似于：

Success
Retry count: 0
Exceptions: []

或在发生异常时：

Function failed after retries: Random KeyError
Retry count: 3
Exceptions: [KeyError('Random KeyError'), KeyError('Random KeyError'), KeyError('Random KeyError')]