Python装饰器进阶：深入理解与最佳实践

1、什么是装饰器

https://docs.python.org/zh-cn/3.7/glossary.html#term-decorator
官网介绍的很简单，返回值为另一个函数的函数，通常使用 @wrapper 语法形式来进行函数变换。装饰器就是闭包的应用，是用来**装饰（修改或增强）**函数、方法、类。

import timedef runtime(function):"""统计运行时间"""def wrapper():"""装饰函数"""start_time = time.time()function()print(f"runtime is {time.time() - start_time}")return wrapperdef fetch_http_data():print('开始请求网络数据')time.sleep(1)print('数据请求完成')@runtime
def parse_response_data():"""解析数据"""print('开始解析数据')time.sleep(0.5)print('数据解析完成')# 把函数当作参数传到另一个函数中执行，但是这种会改变调用方式
decorator = runtime(fetch_http_data)
decorator()# 使用语法糖，不会改变调用方式
parse_response_data()# 被装饰的函数，查看函数名的时候 变成了wrapper，所以装饰器会改变原函数的一些属性
print(parse_response_data.__name__) # wrapper

2、保留装饰器中函数的元数据

parse_response_data.**name** #wrapper
parse_response_data.**doc** # “”“装饰函数”“”
被装饰的函数，查看函数名的时候 变成了wrapper，函数的文档注释也改变了， 所以装饰器会改变原函数的一些属性，如何保留原函数的属性呢？

from functools import wraps# @wrap，它会帮助保留原函数的元信息
# @wraps 有一个重要特征是它能让你通过属性 __wrapped__ 直接访问被包装函数
# parse_response_data.__wrapped__() 可以调用原函数def runtime(function):"""统计运行时间"""@wraps(function)def wrapper():start_time = time.time()function()print(f"runtime is {time.time() - start_time}")return wrapper@runtime
def parse_response_data():print('开始解析数据')time.sleep(0.5)print('数据解析完成')print(parse_response_data.__name__)  # parse_response_data
print(parse_response_data.__doc__)  # 解析数据
# 通过__wrapped__属性调用原函数 
parse_response_data.__wrapped__()

3、带参数的装饰器

装饰器可以带参数，这样可以使装饰器更加灵活和通用，根据不同的情况对被装饰的函数应用不同的行为

def retry(max_retries=3):def decorator(func):def wrapper(*args, **kwargs):for retry_count in range(max_retries):try:result = func(*args, **kwargs)return resultexcept Exception as e:if retry_count < max_retries:print(f"Retrying {func.__name__} (attempt {retry_count + 1}/{max_retries})...")time.sleep(2)else:raise ereturn wrapperreturn decorator@retry(max_retries=2)
def potentially_failing_function():import randomif random.random() < 0.7:print("Function succeeded!")else:raise Exception("Function failed.")potentially_failing_function()

retry 装饰器接受一个 max_retries 参数，用于指定最大的重试次数。decorator 函数接受被装饰的函数 func，并定义了 wrapper 包装函数，该包装函数尝试执行 func，如果遇到异常则进行重试，最多尝试 max_retries 次。
然后，potentially_failing_function 函数应用了带参数的装饰器，允许在最多 2 次重试之后终止或成功执行。

4、 带可选参数的装饰器

import timedef timing_decorator(func=None, message="Execution time"):def decorator(wrapped_func):def wrapper(*args, **kwargs):start_time = time.time()result = wrapped_func(*args, **kwargs)end_time = time.time()execution_time = end_time - start_timeprint(f"{message}: {execution_time} seconds")return resultreturn wrapperif func is None:return decoratorelse:return decorator(func)@timing_decorator(message="Function 1 took")
def function1():time.sleep(2)print("Function 1 completed.")@timing_decorator
def function2():time.sleep(1)print("Function 2 completed.")function1()
function2()

5、 用类实现装饰器

除了使用函数实现装饰器外，类也可以实现装饰器，

import time
import functoolsclass runtime:def __init__(self, func):self.func = func# 保留被装饰函数的元数据functools.update_wrapper(self, func)def __get__(self, instance, owner):if instance is None:return self# 创建一个可调用的对象，将 instance作为self的参数传递进去return functools.partial(self, instance)def __call__(self, *args, **kwargs):start_time = time.time()result = self.func(*args, **kwargs)end_time = time.time()execution_time = end_time - start_timeprint(f"{self.func.__name__} took {execution_time} seconds")return result@runtime
def some_function():time.sleep(2)print("Function completed.")class Animal:@runtimedef walk(self, road):time.sleep(2)print(f"{self.__class__} walk {road}")some_function()
a = Animal()
a.walk('马路')

6、 类中的方法实现装饰器

# 类中的普通方法实现一个装饰器
class DecoratedMeta:def runtime(self, function):"""统计运行时间"""@functools.wraps(function)def wrapper(*args, **kwargs):"""装饰函数"""start_time = time.time()result = function(*args, **kwargs)print(f"runtime is {time.time() - start_time}")return resultreturn wrapper@classmethoddef runtime_cls(cls, function):@functools.wraps(function)def wrapper(*args, **kwargs):print('使用类方法的装饰器')return function(*args, **kwargs)return wrapperd = DecoratedMeta()@d.runtime  # 使用装饰器
def add(x, y):return x + [e-mail]_cls
def sub(x, y):return x - yresult = add(2, 3)
print(result)
result = sub(4, 5)
print(result)

7、 给类加上装饰器

1、给类中的方法加装饰器

import timedef runtime(function):"""统计运行时间"""def wrapper(*args, **kwargs):"""装饰函数"""start_time = time.time()function(*args, **kwargs)print(f"runtime is {time.time() - start_time}")return wrapperclass Animal:@runtimedef walk(self):time.sleep(2)print(f"{self.__class__} walk")a = Animal()
a.walk()

2、给类加装饰器，扩充类的功能

# 定义一个装饰器函数
def log_decorator(cls):# 保存原始类的构造函数original_init = cls.__init__# 定义一个新的构造函数，扩充功能def new_init(self, *args, **kwargs):# 首先调用原始构造函数original_init(self, *args, **kwargs)# 扩展功能：在构造对象时打印信息print(f"创建 {self.__class__.__name__}")# 将新的构造函数替换原始构造函数cls.__init__ = new_initreturn cls# 使用装饰器扩充类的功能
@log_decorator
class MyClass:def __init__(self, x, y):self.x = xself.y = ydef add(self):return self.x * self.y# 创建类的实例
obj = MyClass(3, 4)# 扩充功能生效，构造对象时打印信息
result = obj.add()
print(result)

8、装饰器的叠加

import time
from functools import wrapsdef runtime(function):"""统计运行时间"""@wraps(function)def wrapper(*args, **kwargs):"""装饰函数"""start_time = time.time()result = function(*args, **kwargs)print(f"runtime is {time.time() - start_time}")return resultreturn wrapperdef printlog(function):"""函数运行日志:param function::return:"""@wraps(function)def wrapper(*args, **kwargs):print(f"{function.__name__} start")result = function(*args, **kwargs)print(f"{function.__name__} over")return resultreturn wrapper@printlog
@runtime
def add(x, y):time.sleep(0.5)return x + ydef sub(x, y):return x - y# 调用过程
# a = runtime(add)
# b = printlog(a)
# b(1,3)add(1, 3)a = runtime(sub)
b = printlog(a)
res = b(1, 3)
print(res)

9、 内置的装饰器

@classmethod
把一个方法封装成类方法。

# python的redis第三方库中，使用url连接redis时定义的from_url是一个类方法。
class Redis(object):""""""@classmethoddef from_url(cls, url, db=None, **kwargs):""""""connection_pool = ConnectionPool.from_url(url, db=db, **kwargs)return cls(connection_pool=connection_pool)# 调用类方法
redis_ints = Redis.from_url('redis://user:password@127.0.0.1:6379/0')

@staticmethod
将方法转换为静态方法。

import mathclass CrawlSite:# 使用静态方法计算页数，与实例无关，工具方法@staticmethoddef get_page(total, offsize):"""计算要爬取的页数"""return math.ceil(total / offsize)

@property
会将方法转化为一个具有相同名称的只读属性的 “getter”，特征属性对象具有 getter, setter 以及 deleter 方法，它们可用作装饰器来创建该特征属性的副本，并将相应的访问函数设为所装饰的函数

class Animal(object):def __init__(self, eat):self.__eat = eat# 只有@property时属性不能赋值操作@propertydef eat(self):return [e-mail] eat(self, value):# 设置属性值，同时可以做校验、计算等if not isinstance(value, str):raise TypeError('Expected a string')self.__eat = [e-mail] eat(self):del self.__eata = Animal('rot')
print(a.eat)
a.eat = 'cao'
print(a.eat)

@functools.wraps
保留被装饰的函数元信息，用于在定义包装器函数时发起调用 update_wrapper() 作为函数装饰器
@functools.lru_cache
一个为函数提供缓存功能的装饰器，如果调用相同，则直接返回缓存中的值，不需要重新计算。用以节约高开销或I/O函数的调用时间。
如果 maxsize 设置为 None ，LRU功能将被禁用且缓存数量无上限。由于使用了字典存储缓存，所以该函数的固定参数和关键字参数必须是可哈希的。

@lru_cache(maxsize=100)  
def fibonacci(n):  if n < 2:  return n  return fibonacci(n-1) + fibonacci(n-2)  print(fibonacci(10))  # 输出 55  
print(fibonacci(10))  # 输出 55，不会重新计算

@functools.singledispatch
实现只有第一个参数可接受不同类型的函数

from functools import singledispatch@singledispatch
def calculate_area(argument):raise NotImplementedError('Unsupported operand type')@calculate_area.register(int)
def _(argument):return argument * argument@calculate_area.register(str)
def _(argument):return int(argument) * int(argument)print(calculate_area(5))  # 输出 25
print(calculate_area('6'))  # 输出 36

@contextlib.contextmanager
它可以定义一个支持 with 语句上下文的工厂函数， 而不需要创建一个类或区 enter() 与 exit() 方法。

import contextmanager
import timedef adds():for i in range(3):print(i)time.sleep(1)@contextlib.contextmanager
def timing_context(func):start_time = time.time()try:func()yield 'runtime'  # 进入上下文 yield后面的值，就会赋在 with语句的as 后面finally:end_time = time.time()elapsed_time = end_time - start_timeprint(f"Elapsed time: {elapsed_time} seconds")# 使用上下文管理器来测量代码块的执行时间
with timing_context(adds) as msg:# 模拟耗时操作print(msg)

10、 自定义常用的装饰器

重试机制

import functools
import timedef retries(max_retries=3, delay=1):def decorator(func):@functools.wraps(func)def wrapper(*args, **kwargs):retry_count = 0while retry_count < max_retries:try:return func(*args, **kwargs)except Exception as e:print(f"Error: {func.__name__} failed with {e}. Retrying in {delay} seconds...")retry_count += 1time.sleep(delay)raise Exception(f"Error: {func.__name__} failed after {max_retries} retries.")return wrapperreturn decorator@retries()
def some_function():# Some code that might fail.print('----------------')@retries(max_retries=5, delay=3)
def another_function():# Some code that might fail.print('=============')raisesome_function()another_function()

超时判断

import timeimport functools
from concurrent import futurespool = futures.ThreadPoolExecutor(1)def runtime(seconds):def decorator(func):@functools.wraps(func)def wrapper(*args, **kw):future = pool.submit(func, *args, **kw)return future.result(timeout=seconds)return wrapperreturn decorator@runtime(3)
def request_http():time.sleep(2)return 111res = request_http()
print(res)