Python装饰器是将其他函数添加到另一个函数中的函数，从而可以在不直接更改源代码的情况下添加额外的功能或修改其行为。这是一种特殊的函数，它以另一个函数作为输入，并对其功能进行一些更改。

装饰器在各种情况下都非常有用，比如

• 功能扩展：我们可以通过使用装饰器将日志记录、性能测量或缓存等功能添加到我们的函数中。

• 代码可重用性：我们可以将装饰器甚至类应用于多个实体，可以避免代码重复，并确保不同组件之间的一致行为。

• 行为修改：我们可以拦截输入参数，例如，验证输入变量，而无需许多 assert 行。

 例子

 1 — 计时器

这个装饰器用于计算函数的执行时间并打印出的时间。它对于代码的性能分析和优化非常有用。

import time

def timer(func):
    def wrapper(*args, **kwargs):
        # 记录开始时间
        start_time = time.time()
        # 调用被装饰函数
        result = func(*args, **kwargs)
        # 记录结束时间
        end_time = time.time()
        # 计算执行时间
        execution_time = end_time - start_time
        print(f"Execution time: {execution_time} seconds")
        # 返回被装饰函数的结果
        return result
    # 返回装饰器的引用
    return wrapper

为了在Python中创建装饰器，我们需要定义一个名为 timer 的函数，该函数接受一个名为 func 的参数，以指示它是一个装饰器函数。在计时器函数内部，我们定义另一个名为 wrapper 的函数，该函数接受通常传递给我们想要装饰的函数的参数。

在装饰器内部，我们使用提供的参数调用所需的函数。我们可以使用以下代码行来实现这一点： result = func(*args, **kwargs) 。

最后，装饰器返回装饰函数执行的结果。装饰器函数应该返回对刚刚创建的装饰器的引用。

要使用装饰器，您可以使用 @ 符号将其应用于所需的函数。

@timer
def train_model():
    print("Starting the model training function...")
    # 模拟程序运行时间
    time.sleep(5) 
    print("Model training completed!")

train_model()

 2—调试器

可以创建一个额外的有用的装饰器来方便调试，打印每个函数的输入和输出。这种方法可以让我们了解各个函数的执行流程，而不会在应用程序中弄乱多个打印语句。

def debug(func):
    def wrapper(*args, **kwargs):
        # 打印函数名和参数
        print(f"Calling {func.__name__} with args: {args} kwargs: {kwargs}")
        # 调用函数
        result = func(*args, **kwargs)
        # 打印结果
        print(f"{func.__name__} returned: {result}")
        return result
    return wrapper

我们可以使用 __name__ 参数来获取被调用的函数名称，然后使用 args 、 kwargs 参数来打印传递给函数的内容。

@debug
def add_numbers(x, y):
    return x + y
add_numbers(7, y=5,)  # Output: Calling add_numbers with args: (7) kwargs: {'y': 5} n add_numbers returned: 12

 3 — 异常处理程序

装饰器将捕获在 exception_handler 函数内引发的任何异常，并相应地处理它们。

def exception_handler(func):
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except Exception as e:
            # 处理异常
            print(f"An exception occurred: {str(e)}")
            # 可以添加额外的日志信息
            # you可以重新raise异常
    return wrapper

这对于整理我们的代码并建立统一的异常处理和错误日志记录程序非常有用。

@exception_handler
def divide(x, y):
    result = x / y
    return result
divide(10, 0)  # Output: An exception occurred: division by zero

def validate_input(*validations):
    def decorator(func):
        def wrapper(*args, **kwargs):
            for i, val in enumerate(args):
                if i < len(validations):
                    if not validations[i](val):
                        raise ValueError(f"Invalid argument: {val}")
            for key, val in kwargs.items():
                if key in validations[len(args):]:
                    if not validations[len(args):][key](val):
                        raise ValueError(f"Invalid argument: {key}={val}")
            return func(*args, **kwargs)
        return wrapper
    return decorator

要调用经过验证的输入，我们需要定义验证函数。例如，可以使用两个验证函数。第一个函数（ lambda x: x > 0 ）检查参数 x 是否大于0，第二个函数（ lambda y: isinstance(y, str) ）检查参数 y 是否为字符串类型。

@validate_input(lambda x: x > 0, lambda y: isinstance(y, str))
def divide_and_print(x, message):
    print(message)
    return 1 / x

divide_and_print(5, "Hello!")  # Output: Hello! 1.0

 5 — 重试

这个装饰器会在重试之间延迟一段时间，指定次数重新执行函数。在处理由于临时问题而偶尔失败的网络或API调用时非常有用。

为了实现这一点，我们可以为我们的装饰器定义另一个包装函数，类似于我们之前的示例。然而，这一次我们不再将验证函数作为输入变量提供，而是可以传递特定的参数，如 max_attemps 和 delay 。

import time

def retry(max_attempts, delay=1):
    def decorator(func):
        def wrapper(*args, **kwargs):
            attempts = 0
            while attempts < max_attempts:
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    attempts += 1
                    print(f"Attempt {attempts} failed: {e}")
                    time.sleep(delay)
            print(f"Function failed after {max_attempts} attempts")
        return wrapper
    return decorator

@retry(max_attempts=3, delay=2)
def fetch_data(url):
    print("Fetching the data..")
    # 模拟出现异常
    raise TimeoutError("Server is not responding.")
fetch_data("https://example.com/data")  # 重试3次，以2秒间隔

 结论