经验首页 前端设计 程序设计 Java相关 移动开发 数据库/运维 软件/图像 大数据/云计算 其他经验
当前位置：技术经验 » 程序设计 » Python3 » 查看文章
adorner 使用示例
来源：cnblogs　　作者：顾平安　　时间：2024/7/13 11:15:38　　对本文有异议

模块介绍

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

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

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

安装

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

  1. pip install adorner

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

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

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

类 Decorator

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

源码注释

  1. class Decorator(object):
  2.     def __init__(self, decorator=None):
  3.         self.decorator = decorator or (lambda s: s.execute())  # 是传入的装饰器函数，如果没有传入，则默认使用一个简单的 lambda 函数，该函数调用 self.execute()。
  4.         self.function = None  # 用于存储被装饰的函数。
  5.         # args 和 .kwargs 分别用于存储传递给被装饰函数的位置参数和关键字参数。
  6.         self.args = tuple()
  7.         self.kwargs = dict()
  9.     def __call__(self, function):
  10.         """
  11.         这里是关键部分
  12.         __call__ 可使得类的实例可以像函数一样被调用。
  13.         接收一个函数 function 作为参数，并返回一个 wrapper 函数。
  14.         wrapper 函数内部将被装饰的函数及其参数存储在 self.function、self.args 和 self.kwargs 中，然后调用 self.decorator(self)。
  15.         functools.wraps(function) 用于保持被装饰函数的元数据（如函数名和文档字符串）
  16.         """
  17.         @functools.wraps(function)
  18.         def wrapper(*args, **kwargs):
  19.             self.function = function
  20.             self.args = args
  21.             self.kwargs = kwargs
  22.             return self.decorator(self)
  24.         return wrapper
  26.     def __repr__(self):
  27.         """
  28.         返回对象的字符串表示形式，便于调试和查看对象信息。根据是否有被装饰的函数来返回不同的字符串。
  29.         """
  30.         decorator_name = self.decorator_name.lstrip('<').rstrip('>')
  31.         if self.function:
  32.             return f'<{self.__class__.__name__}: {decorator_name} To {self.function.__name__}>'
  33.         return f'<{self.__class__.__name__}: {decorator_name}>'
  35.     def execute(self, *args, **kwargs):
  36.         """
  37.         用于执行被装饰的函数。会使用传入的参数（如果有）或存储的参数来调用 _execute_sync 方法，该方法应该是为了以后适配更复杂的异步装饰器所提前编写好的
  38.         """
  39.         final_args = args if args else self.args
  40.         final_kwargs = kwargs if kwargs else self.kwargs
  41.         return self._execute_sync(final_args, final_kwargs)
  43.     def _execute_sync(self, args, kwargs):
  44.         """
  45.         同步地执行被装饰的函数，并返回其结果
  46.         """
  47.         return self.function(*args, **kwargs)
  49.     @property
  50.     def function_name(self):
  51.         """返回被装饰函数的名称"""
  52.         if self.function:
  53.             return self.function.__name__
  54.         return '<None>'
  56.     @property
  57.     def function_doc(self):
  58.         """返回被装饰函数的文档字符串"""
  59.         if self.function:
  60.             return self.function.__doc__ or ''
  61.         return ''
  63.     @property
  64.     def decorator_name(self):
  65.         """返回装饰器的名称"""
  66.         if self.decorator:
  67.             return self.decorator.__name__
  68.         return '<None>'
  70.     @property
  71.     def decorator_doc(self):
  72.         """返回装饰器的文档字符串"""
  73.         if self.decorator:
  74.             return self.decorator.__doc__ or ''
  75.         return ''

示例用法

  1. import time
  2. from adorner import Decorator
  5. @Decorator
  6. def exception_decorator(self: Decorator):
  7.     """
  8.     捕获异常日志的装饰器
  9.     :param self: 装饰器 Decorator 实例
  10.     :return: 被修饰函数的执行结果
  11.     """
  12.     print(self.function_doc)  # 打印被装饰函数的文档
  13.     print(self.decorator_doc)  # 打印装饰器的文档
  14.     print(self.function_name)  # 打印被装饰函数的名称
  15.     print(self.decorator_name)  # 打印装饰器的名称
  16.     print(self.args)  # 打印被装饰函数的传入的位置参数 （默认形参值不包含）
  17.     print(self.kwargs)  # 打印被装饰函数的传入的关键字参数  （默认形参值不包含）
  18.     try:
  19.         result = self.execute()  # 打印 1
  20.         # 执行被装饰函数，不传入任何参数时，表示使用默认的参数 self.args、self.kwargs
  21.         # 可覆盖传入参数
  22.         self.execute(value=2)  # 打印 2
  23.         self.execute(3)  # 打印3 并抛出异常
  24.         return result
  25.     except Exception as e:
  26.         print(f"捕获异常: {e}")
  27.         raise
  30. @exception_decorator
  31. def risky_function(value=1):
  32.     print(value)
  33.     if value == 3:
  34.         raise ValueError("出错了")
  37. try:
  38.     risky_function()
  39. except ValueError:
  40.     pass  # 捕获异常: 出错了

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

  3.     捕获异常日志的装饰器
  4.     :param self: 装饰器 Decorator 实例
  5.     :return: 被修饰函数的执行结果
  6.     
  7. risky_function
  8. exception_decorator
  9. ()
  10. {}
  11. 1
  12. 2
  13. 3
  14. 捕获异常: 出错了

类 Timer

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

源码注释

  1. class Timer(Decorator):
  2.     def __init__(self, decorator=None):
  3.         super().__init__(decorator)  # 调用父类 Decorator 的构造函数，初始化装饰器函数。
  4.         self.time = 0  # 用于存储被装饰函数的执行时间。
  6.     def execute(self, *args, **kwargs):
  7.         """
  8.         执行被装饰的函数，并记录其执行时间。
  9.         使用 time.perf_counter() 记录开始和结束的时间，计算函数执行时长，并存储在 self.time 中。
  10.         """
  11.         _start = time.perf_counter()  # 记录开始时间。
  12.         result = super().execute(*args, **kwargs)  # 调用父类的 execute 方法执行被装饰的函数。
  13.         _end = time.perf_counter()  # 记录结束时间。
  14.         self.time = _end - _start  # 计算并存储执行时间。
  15.         return result  # 返回被装饰函数的结果。

示例用法

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

  1. import time
  2. from adorner import Timer
  4. timer = Timer()  # 可装饰多个函数，不过不太推荐（多个函数先后执行会覆盖掉计时器的元数据）
  7. @timer
  8. def my_function(a, b):
  9.     """一个简单的函数，用于演示 Timer 装饰器的使用。"""
  10.     time.sleep(1)  # 模拟一个耗时操作。
  11.     return a + b
  14. result = my_function(1, 2)
  15. print(f'Execution result: {result}')
  16. print(f"Execution time: {timer.time} seconds")

输出将类似于：

  1. Execution result: 3
  2. Execution time: 1.0067455 seconds

类 Cacher

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

源码注释

  1. class Cacher:
  2.     hash = dict()  # 用于存储每个被装饰函数的 Cacher 实例。
  4.     def __new__(cls, function):
  5.         """
  6.         确保每个被装饰的函数只有一个 Cacher 实例。
  7.         如果该函数已经有一个 Cacher 实例，则返回该实例；
  8.         否则，创建一个新的实例，并将其存储在 hash 中。
  9.         """
  10.         if function in cls.hash:
  11.             instance = cls.hash[function]
  12.         else:
  13.             instance = object.__new__(cls)
  14.             instance.function = function  # 设置缓存实例对应的函数
  15.             instance.data = dict()  # 缓存存储的结构是字典
  16.             setattr(instance, '__name__', f'{cls.__name__}-{function.__name__}')
  17.             cls.hash[function] = instance
  19.         return instance
  21.     def __call__(self, *args, **kwargs):
  22.         """
  23.         使 Cacher 实例可以像函数一样被调用。
  24.         调用被装饰的函数，并返回其结果。
  25.         """
  26.         return self.function(*args, **kwargs)
  28.     def __repr__(self):
  29.         """
  30.         返回对象的字符串表示形式，便于调试和查看对象信息。
  31.         """
  32.         return f'<{self.__class__.__name__}: {self.function.__name__}>'
  34.     def __iter__(self):
  35.         """
  36.         使 Cacher 实例可迭代，迭代缓存数据。
  37.         """
  38.         return iter(self.data)
  40.     def __contains__(self, item):
  41.         """
  42.         判断缓存数据中是否包含指定的键。
  43.         """
  44.         return item in self.data
  46.     def __add__(self, other):
  47.         """
  48.         支持使用 + 运算符合并缓存数据。
  49.         """
  50.         if isinstance(other, self.__class__):
  51.             self.data.update(other.data)
  52.             return self
  53.         if isinstance(other, dict):
  54.             self.data.update(other)
  55.             return self
  56.         if isinstance(other, (tuple, list)):
  57.             self.data.update(dict(other))
  58.             return self
  59.         raise TypeError(f'unsupported operand type(s) for +: \'{type(self)}\' and \'{type(other)}\'')
  61.     def __sub__(self, other):
  62.         """
  63.         支持使用 - 运算符从缓存数据中删除指定的键。
  64.         """
  65.         if isinstance(other, self.__class__):
  66.             for key in other.data:
  67.                 self.data.pop(key, None)
  68.             return self
  69.         if isinstance(other, dict):
  70.             for key in other:
  71.                 self.data.pop(key, None)
  72.             return self
  73.         if isinstance(other, (tuple, list)):
  74.             self.pops(*other)
  75.             return self
  76.         raise TypeError(f'unsupported operand type(s) for -: \'{type(self)}\' and \'{type(other)}\'')
  77.     
  78.     def items(self):
  79.         return self.data.items()
  80.     
  81.     def set(self, key, value, safe=False):
  82.         """
  83.         设置缓存数据。
  84.         如果 safe 为 True，则只有在 key 不存在的情况下才设置值。
  85.         """
  86.         if not safe:
  87.             self.data[key] = value
  88.         elif key not in self.data:
  89.             self.data[key] = value
  91.         return self.data[key]
  93.     def sets(self, **data_dict):
  94.         """
  95.         批量设置缓存数据。
  96.         """
  97.         self.data.update(data_dict)
  99.     def get(self, key, default_value=None):
  100.         """
  101.         获取缓存数据。
  102.         如果 key 不存在，则返回 default_value。
  103.         """
  104.         return self.data.get(key, default_value)
  106.     
  107.     @staticmethod
  108.     def _apply_filter(values, filter_function, filter_safe, filter_errors):
  109.         """应用筛选函数"""
  110.         def safe_filter(value):
  111.             try:
  112.                 return filter_function(value)
  113.             except filter_errors:
  114.                 return False
  116.         filter_func = safe_filter if filter_safe else filter_function
  117.         return {key: value for key, value in values.items() if filter_func(value)}
  119.     @staticmethod
  120.     def _apply_map(values, map_function, map_safe, map_errors):
  121.         """应用遍历处理的函数"""
  122.         def safe_map(value_):
  123.             try:
  124.                 return True, map_function(value_)
  125.             except map_errors:
  126.                 return False, None
  128.         if map_safe:
  129.             new_values = {}
  130.             for key, value in values.items():
  131.                 success, mapped_value = safe_map(value)
  132.                 if success:
  133.                     new_values[key] = mapped_value
  134.             return new_values
  135.         else:
  136.             return {key: map_function(value) for key, value in values.items()}
  138.     
  139.     def gets(self, *keys, default_value=None, filter_function=None, map_function=None):
  140.         """
  141.         批量获取缓存数据。
  142.         支持通过 filter_function 过滤值，通过 map_function 处理值。
  143.         """
  144.         values = {key: self.data.get(key, default_value) for key in keys}
  146.         if filter_function:
  147.             filter_errors = filter_errors or (TypeError, ValueError, KeyError, IndexError)
  148.             values = self._apply_filter(values, filter_function, filter_safe, filter_errors)
  150.         if map_function:
  151.             map_errors = map_errors or (TypeError, ValueError, KeyError, IndexError)
  152.             values = self._apply_map(values, map_function, map_safe, map_errors)
  154.         return values
  156.     def pop(self, key, default_value=None):
  157.         """
  158.         删除并返回缓存数据中的指定键。
  159.         如果键不存在，则返回 default_value。
  160.         """
  161.         return self.data.pop(key, default_value)
  163.     def pops(self, *keys, default_value=None):
  164.         """
  165.         批量删除并返回缓存数据中的指定键。
  166.         如果键不存在，则返回 default_value。
  167.         """
  168.         return [self.data.pop(key, default_value) for key in keys]

使用案例

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

  1. from adorner import Cacher
  4. @Cacher
  5. def example1(x):
  6.     """计算乘积"""
  7.     return x * x
  10. @Cacher
  11. def example2(x):
  12.     """计算和"""
  13.     return x + x
  16. print(example1)  # 打印：<Cacher: example>
  17. # 正常调用
  18. print(example1(4))  # 打印：16
  19. # 打印函数的文档字符串
  20. print(example1.function_doc)
  22. # 缓存设置数据
  23. example1.set('a', 1)
  24. example1.set('b', 2)
  25. example1.set('c', 3)
  27. # example2.set('a', True)
  28. # example2.set('b', False)
  29. # 和上述一致
  30. example2.sets(a=True, b=False, d='数据 d')
  32. # 获取缓存数据
  33. print(example1.get('a'))
  34. print(example1.get('d', '数据不存在'))
  35. # 检查 d 是否在缓存器 example1 中
  36. print('d' in example1)
  38. # 缓存数据合并
  39. new_cacher = example1 + example2
  40. print(new_cacher.data)  # 缓存器的所有数据
  41. # 打印：{'a': True, 'b': False, 'c': 3, 'd': '数据 d'}
  43. print(list(new_cacher))  # 将缓存器转为列表，可呈现存储的键
  45. new_cacher += {'e': '合并的数据 e'}
  46. # 迭代打印
  47. for k, v in new_cacher.items():
  48.     print(k, v)
  50. # 批量获取数据
  51. print(new_cacher.gets('a', 'b', 'z', default_value='没有这个数据'))
  52. print(new_cacher.gets('a', 'b', 'c', filter_function=lambda x: x > 1))
  53. # 如果比较类型不一致，可能会发生错误，比如下面这个例子：
  54. # print(new_cacher.gets('a', 'b', 'c', 'd', filter_function=lambda x: x > 1))
  55. # 解决方式：你可以自行捕捉，但是那样会很繁琐，推荐使用 filter_safe 参数
  56. print(new_cacher.gets('a', 'b', 'c', 'd', filter_function=lambda x: x > 1, filter_safe=True))
  57. # 如果启用了 filter_safe 参数还无法正常捕捉，请使用 filter_errors 指定异常，默认是 (TypeError, ValueError, KeyError, IndexError)
  58. print(new_cacher.gets('a', 'b', 'c', 'd', filter_function=lambda x: x(),
  59.                       filter_safe=True, filter_errors=(TypeError, ValueError, KeyError, IndexError)))
  61. # 除了上述的 filter_function 参数，另外还有 map_function，同理也有 map_safe 以及 map_errors 参数
  62. print(new_cacher.gets('a', 'b', 'c', map_function=lambda x: x > 1))
  63. print(new_cacher.gets('a', 'b', 'c', 'd', map_function=lambda x: x > 1, map_safe=True))
  64. print(new_cacher.gets('a', 'b', 'c', 'd', map_function=lambda x: x > 1, map_safe=True, map_errors=(TypeError,)))
  66. # xxx_safe 参数的功能是当传入的函数执行发生异常时对应的一个处理，当出现异常时，该值对应的键值都不应存在于结果中
  67. # 优先级别：正常获取值 -> filter筛选 -> map遍历处理 -> 返回结果
  69. # 弹出某个值
  70. print(new_cacher.pop('c'))
  71. print(new_cacher.pop('c', default_value=None))  # 上面弹出了，这里尝试弹出一个不存在的，将返回 default_value（默认None）
  72. print(new_cacher.pop('c') == new_cacher.pop('c', default_value=None))
  73. print(new_cacher.data)  # {'a': True, 'b': False, 'd': '数据 d', 'e': '合并的数据 e'}
  75. # 批量弹出
  76. print(new_cacher.pops('b', 'c', default_value='不存在'))
  77. print(new_cacher.data)  # {'a': True, 'd': '数据 d', 'e': '合并的数据 e'}
  79. # 减法删除
  80. sub = new_cacher - []  # 支持减去 字典 {'a', 任意值} 以及元组 ('a',)
  81. print(sub.data)  # {'d': '数据 d', 'e': '合并的数据 e'}
  82. print(new_cacher.data)  # {'d': '数据 d', 'e': '合并的数据 e'}

类 Retryer

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

源码注释

  1. from typing import Union, List, Type
  2. import time
  4. class Retryer:
  5.     def __init__(self, max_retry: Union[int] = 3, delay: Union[int] = 0, catches: List[Type[Exception]] = None):
  6.         """
  7.         初始化 Retryer 实例。
  8.         
  9.         :param max_retry: 最大重试次数，默认为 3。
  10.         :param delay: 每次重试之间的延迟时间（秒），默认为 0。
  11.         :param catches: 需要捕获的异常类型列表，默认为空列表。
  12.         """
  13.         self.max_retry = max_retry
  14.         self.delay = delay
  15.         self.catches = catches or []
  16.         self.exceptions = []
  17.         self.count = 0
  19.     def __call__(self, function):
  20.         """使 Retryer 实例可作为装饰器使用。"""
  21.         return Decorator(self.run)(function)
  23.     def run(self, decorator: Decorator):
  24.         """执行重试逻辑。"""
  25.         _catch_exceptions = tuple(self.catches) if self.catches else Exception
  26.         self.exceptions.clear()
  27.         i = 0
  28.         while i <= self.max_retry:
  29.             self.count = i
  30.             try:
  31.                 result = decorator.execute()
  32.             except _catch_exceptions as e:
  33.                 self.exceptions.append(e)
  34.                 i += 1
  35.                 if i <= self.max_retry:
  36.                     time.sleep(self.delay)
  37.                 continue
  38.             else:
  39.                 return result
  40.         raise self.exceptions[-1]

示例用法

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

  1. import random
  2. from adorner import Retryer
  5. # 创建 Retryer 实例，设置捕获的异常类型为 KeyError，当被装饰的函数中出现该错误时将进行重试
  6. retryer = Retryer(catches=[KeyError])
  9. @retryer
  10. def unreliable_function():
  11.     """一个可能会抛出异常的函数，用于演示 Retryer 装饰器的使用"""
  12.     option = random.randint(0, 2)
  13.     if option == 0:
  14.         raise KeyError('Random KeyError')
  15.     elif option == 1:
  16.         raise ValueError('Random ValueError')
  17.     else:
  18.         return "Success"
  21. try:
  22.     result = unreliable_function()
  23.     print(result)
  24. except Exception as e:
  25.     print(f"Function failed after retries: {e}")
  27. # 打印重试次数和捕获的异常
  28. print(f"Retry count: {retryer.count}")
  29. print(f"Exceptions: {retryer.exceptions}")

输出将类似于：

  1. Success
  2. Retry count: 0
  3. Exceptions: []

或在发生异常时：

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

原文链接：https://www.cnblogs.com/gupingan/p/18299851

 友情链接：直通硅谷 　点职佳 　北美留学生论坛

Python3热门文章

  • PyQt-Fluent-Widgets：一个 Fluent Design 风格的组件库
  • Python3读取文件的操作详解
  • 口播神器,基于Edge,微软TTS(text-to-speech)文字转语音免费开源库edge-tts实践(Python3.10)
  • 向量数据库Chroma极简教程
  • 笔精墨妙,妙手丹青,微软开源可视化版本的ChatGPT:Visual ChatGPT,人工智能AI聊天发图片,Python3.10实现
  • win10+vs2019 编译webrtc m108
  • 闻其声而知雅意,基于Pytorch(mps/cpu/cuda)的人工智能AI本地语音识别库Whisper(Python3.10)
  • 靶机练习: Vikings
  • Python3 实现将bytes图片转jpg格式
  • MacBook m1芯片采用miniforge安装python3.9的方法示例

    • Python3推荐文章

  • 分页查询及其拓展应用案例
  • adorner 使用示例
  • 使用pyqt5制作简单计分桌面应用
  • cuda性能优化-2.访存优化
  • 通过钩子函数+Traceid实现Flask链路追踪
  • RCTF 2024 WEB wp
  • 【昇腾开发全流程】AscendCL开发板模型推理
  • PVT：特征金字塔在Vision Transormer的首次应用，又快又好 | ICCV 2021
  • 「网络流浅谈」最小割的模型 - E-Syrus
  • 比Selenium更优秀的playwright介绍与未来展望

    • 本站QQ群：前端 618073944 | Java 606181507 | Python 626812652 | C/C++ 612253063 | 微信 634508462 | 苹果 692586424 | C#/.net 182808419 | PHP 305140648 | 运维 608723728

    W3xue 的所有内容仅供测试，对任何法律问题及风险不承担任何责任。通过使用本站内容随之而来的风险与本站无关。
    关于我们  |  意见建议  |  捐助我们  |  报错有奖  |  广告合作、友情链接（目前9元/月）请联系QQ：27243702 沸活量
    皖ICP备17017327号-2　皖公网安备34020702000426号