模块介绍
adorner 是一个现代轻量级的 Python 装饰器辅助模块。
目前该模块仅实现了 4 个类，对应着 4 个功能：制造装饰器、执行计时、函数缓存、捕获重试。
安装
该模块可在上方仓库中的 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 '<None>' @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 '<None>' @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) # 打印：<Cacher: example> # 正常调用 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')] 
