装饰器

什么是装饰器

饰器是很久以前在Python的PEP-318中引入的，它作为一种机制，用于简化函数和方法的定义，尤其是在需要对已有函数进行修改的情况下。

首先，我们需要理解在Python中，函数就像几乎所有其他东西一样，是常规的对象。这意味着你可以将它们赋值给变量，通过参数传递，甚至将其他函数应用到它们。有时候，我们可能会编写一个小函数，然后对其进行一些变换，生成一个修改后的新函数（这与数学中的函数复合类似）。

例如，如果我们有一个名为original的函数，然后我们有另一个modifier函数，它改变了original的行为，我们需要写如下代码：

def original(...):
    pass
original = modifier(original)

注意这里，我们有了两个可以被调用的original，这样代码会引起一些潜在的问题，比如调用端可能并不清楚调用的是那个original，从而产生一些难以返现的bug。

前面的例子可以被重写如下：

@modifier
def original(...):
   pass

可以说装饰器只是语法糖，用于将装饰器后面的内容作为装饰器本身的第一个参数进行调用，结果就是装饰器返回的内容。装饰器的语法显著提高了代码的可读性，因为现在阅读代码的人可以在一个地方找到函数的整个定义。请注意，像以前那样手动修改函数仍然是允许的。

最佳实践

一般来讲，我们应该避免在不使用装饰器语法的情况下，对已经设计好的函数进行重新赋值，特别是在远离函数原始定义位置的地方，这将极大地降低代码的可读性。

按照Python的术语，以及我们的示例，modifier是我们所称的装饰器，而original是被装饰的函数，通常也被称为封装对象。

虽然这个功能最初是为方法和函数设计的，但实际的语法允许任何类型的对象被装饰，所以我们将探讨应用于函数、方法、生成器和类的装饰器。

最后要注意的是，虽然装饰器这个名称是正确的（毕竟，装饰器正在对被封装的函数进行修改、扩展或协同工作），但不要与装饰器设计模式混淆。

函数装饰器

函数可能是可以被装饰的Python对象中最简单的表示形式。我们可以在函数上使用装饰器来应用各种逻辑——我们可以验证参数，检查前提条件，完全改变行为，修改其签名，缓存结果（创建原始函数的备忘录版本）等等。

作为一个例子，我们将创建一个基本的装饰器，实现一个重试机制，用于控制特定的域级别异常，并尝试重试一定次数：

from functools import wraps

def retry(operation):
    @wraps(operation)
    def wrapped(*args, **kwargs):
        last_raised = None
        RETRIES_LIMIT = 3
        for _ in range(RETRIES_LIMIT):
            try:
                return operation(*args, **kwargs)
            except Exception as e:
                print("retrying...")
                last_raised = e
        raise last_raised
    return wrapped

@retry
def faulty_operation():
    raise Exception("Exception occurred")

if __name__ == "__main__":
    try:
        faulty_operation()
    except Exception:
        print("Operation failed after 3 retries.")

运行结果为：

retrying...
retrying...
retrying...
Operation failed after 3 retries.

习惯用法

在for循环中使用_意味着数据被赋值给一个我们当前并不关心的变量，因为它并未在for循环内部使用（在Python中，通常用_命名那些被忽略的值，这是一种常见的习惯用法）。

类装饰器

类装饰器在PEP-3129中被引入，它与我们刚刚探讨的函数装饰器非常类似。唯一的区别是我们正在接收一个类作为被包装方法的参数，而不是另一个函数。

一个经典的内建类装饰器，是dataclasses.dataclass装饰器。在本章中，我们将学习如何编写我们自己的类装饰器。

一些观点可能会认为，装饰一个类是一种相当复杂的事情，可能会危害到代码的可读性，因为我们将在类中声明一些属性和方法，但在幕后，装饰器可能正在修改他们，这将产生一个完全不同的类。

这种评估是正确的，但只有在严重滥用这种技术的情况下才是如此。客观地说，这与装饰函数没有什么不同；毕竟，类只是Python生态系统中的另一种类型的对象，就像函数一样。我们将在以后讨论它的利弊，但现在，将探讨类的装饰器的好处：

• 重用代码和遵循DRY（Don’t Repeat Yourself，不要重复自己）。一个有效的类装饰器，可以应用到多个类上，使其符合某种接口或标准（通过装饰器中进行一些检查）。
• 我们可以创建更小或更简单的类，这些类将在后续被装饰器增强。
• 相比于同样可修改类行为的元类（并不推荐），如果我们使用装饰器，会让类的转换逻辑将更易于维护。

考虑如下代码：

class UserLoginEventFormatter:
    def __init__(self, event):
        self.event = event

    def format(self) -> dict:
        return {
            "username": self.event.username,
            "password": "**redacted**",
            "ip": self.event.ip,
            "timestamp": self.event.timestamp.strftime("%Y-%m-%d %H:%M"),
        }

@dataclass
class UserLoginEvent:
    FORMATTER = UserLoginEventFormatter
    username: str
    password: str
    ip: str
    timestamp: datetime

    def format(self) -> dict:
        return self.FORMATTER(self).format()

在这里，我们定义了一个类，它直接映射用户登录事件，并包含对应的逻辑：隐藏密码字段，并按需格式化时间戳。

虽然这个方法可行，一开始可能看起来是个不错的选择，但随着时间的推移，当我们想要扩展我们的系统时，我们会遇到一些问题：

1. 类的数量暴增：随着事件数量的增长，格式化类的数量也会以相同的幅度增长，因为它们是一对一映射的。

2. 解决方案不够灵活：如果我们需要重用部分组件（例如，我们需要在另一种类型的事件中隐藏密码），我们将不得不将其提取到一个函数中，但同时又需要在多个类中反复调用它，这意味着我们并没有最大化地重用代码。

3. 代码冗余：format()方法将需要出现在所有的事件类中。尽管我们可以将这部分代码提取到另一个类中（创建一个混入类，mixin），但这并不是继承的最佳实践方式。

   混入类是一种提供某种特定功能，但不是为了自身实例化的类。它们被设计为其他类的组成部分，可以通过多重继承被添加到其他类中以提供某种特定功能。在这个例子中，一个混入类可能会包含format()方法，然后其他所有的事件类都会继承这个混入类。

   然而，这种方法可能并不是继承的最佳使用方式。在面向对象编程中，继承主要应被用于表示”是一个”（is-a）关系，例如一个Dog类可以继承自Animal类，因为狗是动物。而在这个例子中，我们的事件类并不是格式化类，因此使用继承可能并不合适。此外，过度使用继承也可能导致代码的结构变得复杂，而使用装饰器可以更加清晰地表达代码的意图。

另一种解决方案是动态地构建一个对象，给定一组过滤器（转换函数）和一个事件实例，它可以通过将过滤器应用到其字段上来格式化它。然后，我们只需要定义函数来转换每种类型的字段，格式化器是通过组合许多这样的函数来创建的。

这其实是一种函数式编程的思想，我们通过组合多个函数（即转换函数）来动态创建一个格式化器对象。这个格式化器对象接受一组过滤器（在这里，过滤器实际上就是一组预定义好的函数，它们负责对事件实例的字段进行特定的转换操作）和一个事件实例，然后它将这些过滤器应用到事件实例的字段上，最终生成一个格式化后的对象。这种方法的优势在于它的灵活性和可复用性，因为我们只需要定义一次转换函数，就可以在多个不同的格式化器中重复使用它们。

from dataclasses import dataclass
from datetime import datetime

def mask_sensitive_info(field) -> str:
    return "**redacted**"

def format_datetime(field_timestamp: datetime) -> str:
    return field_timestamp.strftime("%Y-%m-%d %H:%M")

def display_as_is(event_field):
    return event_field

class EventDataFormatter:
    def __init__(self, format_rules: dict) -> None:
        self.format_rules = format_rules

    def format(self, event) -> dict:
        return {
            field: rule(getattr(event, field, None))
            for field, rule
            in self.format_rules.items()
        }

class Formatter:
    def __init__(self, **rules):
        self.data_formatter = EventDataFormatter(rules)

    def __call__(self, event_class):
        def format_method(event_instance):
            return self.data_formatter.format(event_instance)

        event_class.format = format_method
        return event_class

@Formatter(
    username=str.lower,
    password=mask_sensitive_info,
    ip=display_as_is,
    timestamp=format_datetime,
)
@dataclass
class UserLoginEvent:
    username: str
    password: str
    ip: str
    timestamp: datetime

# Create an instance of the UserLoginEvent
login = UserLoginEvent('UserName', 'UserPassword', '192.168.1.1', datetime.now())

# Use the added format method to format the login event
formatted_data = login.format()

# Print the formatted data
print(formatted_data)

生成器装饰器

【TODO】

传递参数给装饰器

装饰器是Python中的强大工具。然而，如果我们能够给它们传递参数，使它们的逻辑进一步抽象，它们就能变得更强大。

有几种实现可以接受参数的装饰器的方法，这里只讨论最常见的实现手段。

• 第一种是创建装饰器作为具有新的间接层的嵌套函数，使装饰器中的所有东西都下降一个层级。
• 第二种方法是使用类进行装饰。

一般来说，第二种方法更有利于可读性，因为考虑一个对象比处理闭包的三个或更多的嵌套函数更容易。然而，为了完整性，我们将探讨这两种方法，你可以决定哪种方法最适合手头的问题。

嵌套函数的装饰器

简单来说，装饰器的一般概念是创建一个返回另一个函数的函数（在函数式编程中，接受其他函数作为参数的函数被称为高阶函数，这就是我们在这里讨论的概念）。在装饰器主体中定义的内部函数将是要被调用的那个。

现在，如果我们想要向其传递参数，我们就需要另一层间接性。第一个函数会接受参数，在那个函数内部，我们将定义一个新的函数，这将是装饰器，这又将定义另一个新的函数，即作为装饰过程的结果返回的函数。这意味着我们将至少有三个层次的嵌套函数。

如果到目前为止这一点还不清楚，不用担心。在查看即将来临的示例之后，一切都会变得清晰。

现在，我们希望能够指示每个实例将重试几次，我们也可以为这个参数添加一个默认值。为了做到这一点，我们需要另一层嵌套函数：首先是参数，然后是装饰器本身。

这是因为我们现在将要形成如下形式的代码，作为对比，可以参考最开始没有参数的函数装饰器：

_DEFAULT_RETRIES_LIMIT = 3

def with_retry(
    retries_limit: int = _DEFAULT_RETRIES_LIMIT
):    
    def retry(operation):
        @wraps(operation)
        def wrapped(*args, **kwargs):
            last_raised = None
            for _ in range(retries_limit):
                try:
                    return operation(*args, **kwargs)
                except Exception as e:
                    print("retrying...")
                    last_raised = e
            raise last_raised
        return wrapped
    return retry

整体上可以描述为

<original_function> = with_retry(retries_limit=3)(<original_function>)

注意

with_retry(retries_limit=3)返回的对象是retry()函数，with_retry(retries_limit=3)的目的是接受参数并将参数传递给装饰器函数。

而对应的没有参数的装饰器，整体上可以表述为

<original_function> = retry(<original_function>)

默认参数

在上面的例子中，如果考虑使用默认值，在使用装饰器时可能会产生一些困惑

@with_retry()
def my function(): ...

或是使用

@with_retry
def my function(): ...

如果装饰器接受的参数没有默认值，那么第二种语法就没有意义，只有一种可能性。如果有默认值，那么情况会稍微复杂一些，根据上面的例子，应该使用第一种语法，

另外，你也可以让装饰器同时支持两种语法。这会增加代码的复杂度，你应该权衡这样做是否值得。

我们用一个简单的例子来说明这一点，该例子使用带参数的装饰器将参数注入到函数中。我们定义了一个接受两个参数的函数和一个同样接受两个参数的装饰器，目标是在不使用参数调用函数的情况下让它使用装饰器传递的参数进行工作：

• 使用装饰器的默认参数
• 可以使用partical()函数
• lamba实现

from functools import wraps

DEFAULT_X = 10
DEFAULT_Y = 20

def decorator(function=None, *, x=DEFAULT_X, y=DEFAULT_Y):
    def decorated(function):
        @wraps(function)
        def wrapped():
            return function(x, y)
        return wrapped

    if function:
        return decorated(function)
    return decorated

# This function will be called with DEFAULT_X and DEFAULT_Y
@decorator
def add(x, y):
    return x + y

# This function will be called with 5 and 3
@decorator(x=5, y=3)
def subtract(x, y):
    return x - y

print(add())  # Output: 30
print(subtract())  # Output: 2

def decorator(function=None, *, x=DEFAULT_X, y=DEFAULT_Y):
    if function is None:
        return partial(decorator, x=x, y=y)

    @wraps(function)
    def wrapped():
        return function(x, y)

    return wrapped

def decorator(function=None, *, x=DEFAULT_X, y=DEFAULT_Y):
    if function is None:
        return lambda func: decorator(func, x=x, y=y)

    @wraps(function)
    def wrapped():
        return function(x, y)

    return wrapped

习惯用法

在函数签名中的星号(*)用来表示其后的参数应被视为仅限关键字参数。这意味着这些参数只能通过关键字语法提供，而不能通过位置语法。
以下是一个简单的例子进行演示

def func(a, *, b):
    print(a, b)

func(1, 2)      # Raises a TypeError
func(1, b=2)    # Prints: 1 2

习惯用法

相应的如果在函数定义中看到了/，那么它前面的参数只能作为位置参数使用，不能通过关键字来调用。这有助于使函数的使用方式更加明确，防止误用。

def func(a, /):
    print(a, b)

func(10)  # This is correct Prints: 10
func(a=10)  # This will raise a TypeError

装饰器对象

前面的例子需要三层嵌套函数。第一个函数会接收我们想要使用的装饰器的参数。在这个函数里，其他函数都是闭包函数，可以使用这些参数以及装饰器的逻辑。

一个更为简洁的实现方式是使用类来定义装饰器。在这种情况下，我们可以在__init__方法中传递参数，然后在名为__call__的魔术方法中实现装饰器的逻辑。

装饰器的代码会像下面的例子一样：

from functools import wraps
from typing import Callable

class Retry:
    def __init__(self, retries_limit: int = 3):
        self.retries_limit = retries_limit

    def __call__(self, operation: Callable):
        @wraps(operation)
        def wrapped(*args, **kwargs):
            last_raised = None
            for _ in range(self.retries_limit):
                try:
                    return operation(*args, **kwargs)
                except Exception as e:
                    print("retrying...")
                    last_raised = e
            raise last_raised
        return wrapped

@Retry(retries_limit=3)
def faulty_operation():
    raise Exception("Exception occurred")

if __name__ == "__main__":
    try:
        faulty_operation()
    except Exception:
        print("Operation failed after 3 retries.")

值得注意的是Python语法在这里的作用方式。

1. 创建了装饰器对象Retry，所以在@操作被应用之前，次实例已经创建并传递了参数给它。这将根据在__init__方法中的定义创建一个新对象并使用这些参数进行初始化。
2. @操作被调用，所以这个装饰器对象Retry将会包裹faulty_operation函数，意味着它将被传递给__call__魔法方法。
3. 在__call__魔法方法内部，我们定义了装饰器的逻辑，就像我们通常做的那样——我们包裹了原始函数，返回一个带有我们期望的逻辑的新函数。

协程的装饰器

正如引言中所解释的，由于Python中几乎所有的东西都是对象，因此几乎任何东西都可以被装饰，包括协程。

然而，这里有一个需要注意的地方，那就是，如前面章节中解释的，Python中的异步编程引入了一些语法差异。因此，这些语法差异也将带入装饰器。

简单来说，如果我们要为一个协程写一个装饰器，我们可以简单地适应新的语法（记住要等待被装饰的协程，并将包装的对象定义为一个协程，这意味着内部函数可能需要使用’async def’而不是简单的’def’）。

问题在于，如果我们想要一个既适用于函数又适用于协程的装饰器。在大多数情况下，创建两个装饰器会是最简单（也许也是最好）的方法，但如果我们想要为我们的用户提供一个更简单的接口（通过减少需要记住的对象），我们可以创建一个轻量级的包装器，充当两个内部（不公开）装饰器的调度员。这就像是创建一个装饰器的外观。

关于创建一个适用于函数和协程的装饰器有多困难，并没有通用的规则，因为这取决于我们想要在装饰器本身中添加什么逻辑。例如，在下面的代码中，有一个装饰器可以修改它接收的函数的参数，这既适用于常规函数，也适用于协程：

• 装饰器不需要使用await
• 需要在装饰器内部等待

装饰器将接收协程作为其可调用参数，然后传入参数。这将创建协程对象（将进入事件循环的任务），但装饰器不需要等待，这意味着谁调用await coro()，谁就会最终等待装饰器包装的协程的结果。

X, Y = 1, 2
 
def decorator(callable):
    @wraps(callable)
    def wrapped():
        return callable(X, Y)
 
    return wrapped
 
@decorator
def func(x, y):
    return x + y
 
@decorator
async def coro(x, y):
    return x + y

如果我们需要一个计时功能，那么我们必须等待函数或协程完成以便测量时间，为此我们将必须在其上调用await，这意味着包装对象将反过来必须是一个协程（但主装饰器不必是）。

下面的代码通过一个装饰器示例说明了这一点，该装饰器可以选择性地决定如何包装调用者

import time
import inspect
from functools import wraps
import asyncio


def timing(callable):
    if inspect.iscoroutinefunction(callable):
        async def wrapped_coro(*args, **kwargs):
            start = time.time()
            result = await callable(*args, **kwargs)
            latency = time.time() - start
            return {"latency": latency, "result": result}
        return wrapped_coro

    else:
        @wraps(callable)
        def wrapped(*args, **kwargs):
            start = time.time()
            result = callable(*args, **kwargs)
            latency = time.time() - start
            return {"latency": latency, "result": result}
        return wrapped

@timing
def sync_function():
    for _ in range(10000000):  # simulate a CPU-bound operation
        pass

@timing
async def async_function():
    await asyncio.sleep(1)  # simulate an IO-bound operation

if __name__ == "__main__":
    print(sync_function())
    # Output: {'latency': 0.34108567237854004, 'result': None}

    print(asyncio.run(async_function()))
    # Output: {'latency': 1.0015132427215576, 'result': None}

习惯用法

作为一条通用规则，你应该用同类型的对象替换被装饰的对象，也就是说，用一个函数替换一个函数，用另一个协程替换一个协程。