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 '<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')]

版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。

相关推荐


我最近重新拾起了计算机视觉,借助Python的opencv还有face_recognition库写了个简单的图像识别demo,额外定制了一些内容,原本想打包成exe然后发给朋友,不过在这当中遇到了许多小问题,都解决了,记录一下踩过的坑。 1、Pyinstaller打包过程当中出现warning,跟d
说到Pooling,相信学习过CNN的朋友们都不会感到陌生。Pooling在中文当中的意思是“池化”,在神经网络当中非常常见,通常用的比较多的一种是Max Pooling,具体操作如下图: 结合图像理解,相信你也会大概明白其中的本意。不过Pooling并不是只可以选取2x2的窗口大小,即便是3x3,
记得大一学Python的时候,有一个题目是判断一个数是否是复数。当时觉得比较复杂不好写,就琢磨了一个偷懒的好办法,用异常处理的手段便可以大大程度帮助你简短代码(偷懒)。以下是判断整数和复数的两段小代码: 相信看到这里,你也有所顿悟,能拓展出更多有意思的方法~
文章目录 3 直方图Histogramplot1. 基本直方图的绘制 Basic histogram2. 数据分布与密度信息显示 Control rug and density on seaborn histogram3. 带箱形图的直方图 Histogram with a boxplot on t
文章目录 5 小提琴图Violinplot1. 基础小提琴图绘制 Basic violinplot2. 小提琴图样式自定义 Custom seaborn violinplot3. 小提琴图颜色自定义 Control color of seaborn violinplot4. 分组小提琴图 Group
文章目录 4 核密度图Densityplot1. 基础核密度图绘制 Basic density plot2. 核密度图的区间控制 Control bandwidth of density plot3. 多个变量的核密度图绘制 Density plot of several variables4. 边
首先 import tensorflow as tf tf.argmax(tenso,n)函数会返回tensor中参数指定的维度中的最大值的索引或者向量。当tensor为矩阵返回向量,tensor为向量返回索引号。其中n表示具体参数的维度。 以实际例子为说明: import tensorflow a
seaborn学习笔记章节 seaborn是一个基于matplotlib的Python数据可视化库。seaborn是matplotlib的高级封装,可以绘制有吸引力且信息丰富的统计图形。相对于matplotlib,seaborn语法更简洁,两者关系类似于numpy和pandas之间的关系,seabo
Python ConfigParser教程显示了如何使用ConfigParser在Python中使用配置文件。 文章目录 1 介绍1.1 Python ConfigParser读取文件1.2 Python ConfigParser中的节1.3 Python ConfigParser从字符串中读取数据
1. 处理Excel 电子表格笔记(第12章)(代码下载) 本文主要介绍openpyxl 的2.5.12版处理excel电子表格,原书是2.1.4 版,OpenPyXL 团队会经常发布新版本。不过不用担心,新版本应该在相当长的时间内向后兼容。如果你有新版本,想看看它提供了什么新功能,可以查看Open
1. 发送电子邮件和短信笔记(第16章)(代码下载) 1.1 发送电子邮件 简单邮件传输协议(SMTP)是用于发送电子邮件的协议。SMTP 规定电子邮件应该如何格式化、加密、在邮件服务器之间传递,以及在你点击发送后,计算机要处理的所有其他细节。。但是,你并不需要知道这些技术细节,因为Python 的
文章目录 12 绘图实例(4) Drawing example(4)1. Scatterplot with varying point sizes and hues(relplot)2. Scatterplot with categorical variables(swarmplot)3. Scat
文章目录 10 绘图实例(2) Drawing example(2)1. Grouped violinplots with split violins(violinplot)2. Annotated heatmaps(heatmap)3. Hexbin plot with marginal dist
文章目录 9 绘图实例(1) Drawing example(1)1. Anscombe’s quartet(lmplot)2. Color palette choices(barplot)3. Different cubehelix palettes(kdeplot)4. Distribution
Python装饰器教程展示了如何在Python中使用装饰器基本功能。 文章目录 1 使用教程1.1 Python装饰器简单示例1.2 带@符号的Python装饰器1.3 用参数修饰函数1.4 Python装饰器修改数据1.5 Python多层装饰器1.6 Python装饰器计时示例 2 参考 1 使
1. 用GUI 自动化控制键盘和鼠标第18章 (代码下载) pyautogui模块可以向Windows、OS X 和Linux 发送虚拟按键和鼠标点击。根据使用的操作系统,在安装pyautogui之前,可能需要安装一些其他模块。 Windows: 不需要安装其他模块。OS X: sudo pip3
文章目录 生成文件目录结构多图合并找出文件夹中相似图像 生成文件目录结构 生成文件夹或文件的目录结构,并保存结果。可选是否滤除目录,特定文件以及可以设定最大查找文件结构深度。效果如下: root:[z:/] |--a.py |--image | |--cat1.jpg | |--cat2.jpg |
文章目录 VENN DIAGRAM(维恩图)1. 具有2个分组的基本的维恩图 Venn diagram with 2 groups2. 具有3个组的基本维恩图 Venn diagram with 3 groups3. 自定义维恩图 Custom Venn diagram4. 精致的维恩图 Elabo
mxnet60分钟入门Gluon教程代码下载,适合做过深度学习的人使用。入门教程地址: https://beta.mxnet.io/guide/getting-started/crash-course/index.html mxnet安装方法:pip install mxnet 1 在mxnet中使
文章目录 1 安装2 快速入门2.1 基本用法2.2 输出图像格式2.3 图像style设置2.4 属性2.5 子图和聚类 3 实例4 如何进一步使用python graphviz Graphviz是一款能够自动排版的流程图绘图软件。python graphviz则是graphviz的python实