久趣下载站

当前位置: 首页 » 游戏攻略 » adorner 使用示例

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 '<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')]
猜你喜欢
本类排行