模块介绍
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
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
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
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
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')]
猜你喜欢
