醉了，面个功能测试，还问我Python装饰器

Python 装饰器是个强大的工具，可帮你生成整洁、可重用和可维护的代码。某种意义上说，会不会用装饰器是区分新手和老鸟的重要标志。如果你不熟悉装饰器，你可以将它们视为将函数作为输入并在不改变其主要用途的情况下扩展其功能的函数。装饰器可以有效提高你的工作效率并避免重复代码。本文我整理了项目中经常用到的 12 个装饰器，值得每一个Python开发者掌握。

01 @logger

我们从最简单的装饰器开始，手动实现一个可以记录函数开始和结束的装饰器。

被修饰函数的输出结果如下所示：

some_function(args)
# ----- some_function: start -----
# some_function executing
# ----- some_function: end -----

要实现一个装饰器，首先要给装饰器起一个合适的名称：这里我们给装饰器起名为logger。

装饰器本质上是一个函数，它将一个函数作为输入并返回一个函数作为输出。输出函数通常是输入的扩展版。在我们的例子中，我们希望输出函数用start和end语句包围输入函数的调用。

由于我们不知道输入函数都带有什么参数，我们可以使用 *args 和 **kwargs 从包装函数传递它们。*args 和 **kwargs 允许传递任意数量的位置参数和关键字参数。

下面是logger装饰器的示例代码：

def logger(function):
def wrapper(*args, **kwargs):
print(f"----- {function.__name__}: start -----")
output = function(*args, **kwargs)
print(f"----- {function.__name__}: end -----")
return output
return wrapper

logger函数可以应用于任意函数，比如：

decorated_function = logger(some_function)

上面的语句是正确的，但Python 提供了更 Pythonic 的语法——使用 @ 修饰符。

因此更通常的写法是：

@logger
def some_function(text):
print(text)
some_function("first test")
# ----- some_function: start -----
# first test
# ----- some_function: end -----
some_function("second test")
# ----- some_function: start -----
# second test
# ----- some_function: end -----

02 @wraps

要了解 @wraps 的作用以及为什么需要它，让我们将前面写的logger装饰器应用到一个将两个数字相加的简单函数中。

下面的代码是未使用@wraps装饰器的版本：

def logger(function):
def wrapper(*args, **kwargs):
"""wrapper documentation"""
print(f"----- {function.__name__}: start -----")
output = function(*args, **kwargs)
print(f"----- {function.__name__}: end -----")
return output
return wrapper
@logger
def add_two_numbers(a, b):
"""this function adds two numbers"""
return a + b

如果我们用__name__ 和 __doc__来查看被装饰函数add_two_numbers的名称和文档，会得到如下结果

add_two_numbers.__name__
'wrapper'
add_two_numbers.__doc__
'wrapper documentation'

输出的是wrapper函数的名称和文档。这是我们预期想要的结果，我们希望保留原始函数的名称和文档。这时@wraps装饰器就派上用场了。

我们唯一需要做的就是给wrapper函数加上@wraps装饰器。

from functools import wraps
def logger(function):
@wraps(function)
def wrapper(*args, **kwargs):
"""wrapper documentation"""
print(f"----- {function.__name__}: start -----")
output = function(*args, **kwargs)
print(f"----- {function.__name__}: end -----")
return output
return wrapper
@logger
def add_two_numbers(a, b):
"""this function adds two numbers"""
return a + b

再此检查add_two_numbers函数的名称和文档，我们可以看到该函数的元数据。

add_two_numbers.__name__
# 'add_two_numbers'
add_two_numbers.__doc__
# 'this function adds two numbers'

03 @lru_cache

@lru_cache是Python内置装饰器，可以通过from functools import lru_cache引入。@lru_cache的作用是缓存函数的返回值，当缓存装满时，使用least-recently-used（LRU）算法丢弃最少使用的值。

@lru_cache装饰器适合用于输入输出不变且运行时间较长的任务，例如查询数据库、请求静态页面或一些繁重的处理。

在下面的示例中，我使用@lru_cache来修饰一个模拟某些处理的函数。然后连续多次对同一输入应用该函数。

import random
import time
from functools import lru_cache
@lru_cache(maxsize=None)
def heavy_processing(n):
sleep_time = n + random.random()
time.sleep(sleep_time)
# 初次调用
%%time
heavy_processing(0)
# CPU times: user 363 µs, sys: 727 µs, total: 1.09 ms
# Wall time: 694 ms
# 第二次调用
%%time
heavy_processing(0)
# CPU times: user 4 µs, sys: 0 ns, total: 4 µs
# Wall time: 8.11 µs
# 第三次调用
%%time
heavy_processing(0)
# CPU times: user 5 µs, sys: 1 µs, total: 6 µs
# Wall time: 7.15 µs

从上面的输出可以看到，第一次调用花费了694ms，因为执行了time.sleep()函数。后面两次调用由于参数相同，直接返回缓存值，因此并没有实际执行函数内容，因此非常快地得到函数返回。

04 @repeat

该装饰器的所用是多次调用被修饰函数。这对于调试、压力测试或自动化多个重复任务非常有用。

跟前面的装饰器不同，@repeat接受一个输入参数，

def repeat(number_of_times):
def decorate(func):
@wraps(func)
def wrapper(*args, **kwargs):
for _ in range(number_of_times):
func(*args, **kwargs)
return wrapper
return decorate

上面的代码定义了一个名为repeat的装饰器，有一个输入参数number_of_times。与前面的案例不同，这里需要decorate函数来传递被修饰函数。然后，装饰器定义一个名为wrapper的函数来扩展被修饰函数。

@repeat(5)
def hello_world():
print("hello world")
hello_world()
# hello world
# hello world
# hello world
# hello world
# hello world

05 @timeit

该装饰器用来测量函数的执行时间并打印出来。这对调试和监控非常有用。

在下面的代码片段中，@timeit装饰器测量process_data函数的执行时间，并以秒为单位打印所用的时间。

import time
from functools import wraps
def timeit(func):
@wraps(func)
def wrapper(*args, **kwargs):
start = time.perf_counter()
result = func(*args, **kwargs)
end = time.perf_counter()
print(f'{func.__name__} took {end - start:.6f} seconds to complete')
return result
return wrapper
@timeit
def process_data():
time.sleep(1)
process_data()
# process_data took 1.000012 seconds to complete

06 @retry

其工作原理如下：

wrapper函数启动num_retrys次迭代的for循环。
将被修饰函数放到try/except块中。每次迭代如果调用成功，则中断循环并返回结果。否则，休眠sleep_time秒后继续下一次迭代。
当for循环结束后函数调用依然不成功，则抛出异常。

示例代码如下：

import random
import time
from functools import wraps
def retry(num_retries, exception_to_check, sleep_time=0):
"""
遇到异常尝试重新执行装饰器
"""
def decorate(func):
@wraps(func)
def wrapper(*args, **kwargs):
for i in range(1, num_retries+1):
try:
return func(*args, **kwargs)
except exception_to_check as e:
print(f"{func.__name__} raised {e.__class__.__name__}. Retrying...")
if i < num_retries:
time.sleep(sleep_time)
# 尝试多次后仍不成功则抛出异常
raise e
return wrapper
return decorate
@retry(num_retries=3, exception_to_check=ValueError, sleep_time=1)
def random_value():
value = random.randint(1, 5)
if value == 3:
raise ValueError("Value cannot be 3")
return value
random_value()
# random_value raised ValueError. Retrying...
# 1
random_value()
# 5

07 @countcall

@countcall用于统计被修饰函数的调用次数。这里的调用次数会缓存在wraps的count属性中。

from functools import wraps
def countcall(func):
@wraps(func)
def wrapper(*args, **kwargs):
wrapper.count += 1
result = func(*args, **kwargs)
print(f'{func.__name__} has been called {wrapper.count} times')
return result
wrapper.count = 0
return wrapper
@countcall
def process_data():
pass
process_data()
process_data has been called 1 times
process_data()
process_data has been called 2 times
process_data()
process_data has been called 3 times

08 @rate_limited

@rate_limited装饰器会在被修饰函数调用太频繁时，休眠一段时间，从而限制函数的调用速度。这在模拟、爬虫、接口调用防过载等场景下非常有用

import time
from functools import wraps
def rate_limited(max_per_second):
min_interval = 1.0 / float(max_per_second)
def decorate(func):
last_time_called = [0.0]
@wraps(func)
def rate_limited_function(*args, **kargs):
elapsed = time.perf_counter() - last_time_called[0]
left_to_wait = min_interval - elapsed
if left_to_wait > 0:
time.sleep(left_to_wait)
ret = func(*args, **kargs)
last_time_called[0] = time.perf_counter()
return ret
return rate_limited_function
return decorate

该装饰器的工作原理是：测量自上次函数调用以来所经过的时间，并在必要时等待适当的时间，以确保不超过速率限制。其中等待时间=min_interval - elapsed，这里min_intervalue是两次函数调用之间的最小时间间隔（以秒为单位），已用时间是自上次调用以来所用的时间。如果经过的时间小于最小间隔，则函数在再次执行之前等待left_to_wait秒。