如何使用装饰器来修改多个函数的行为？

Question

我正在学习 Python 中的装饰器，并希望使用它们来修改代码中几个更复杂函数的行为。具体来说，我想：

• 记录每个函数的执行时间。
• 处理并记录函数执行期间发生的任何错误。

这是我的一个函数的示例，它处理数字列表，过滤掉偶数，并计算剩余奇数的总和：

def my_function(numbers):
    odd_numbers = [n for n in numbers if n % 2 != 0]
    result = sum(odd_numbers)
    return result

我还有另一个函数，通过反转句子中的单词来处理字符串：

def reverse_sentence(sentence):
    words = sentence.split()
    reversed_sentence = ' '.join(reversed(words))
    return reversed_sentence

我了解如何创建一个简单的装饰器，但是如何创建一个具有以下功能的装饰器：

• 记录每个函数的执行时间。
• 捕获并记录函数执行期间发生的任何错误。

我想将此装饰器应用于这两个函数（以及可能的其他函数）。任何有关示例的帮助将不胜感激！

Answer 1

也许这个例子会有所帮助：

import time

def measure_function(f):
    def execute(*args, **kwargs):
        t = time.time()
        try:
            res = f(*args, **kwargs)
            print("function %s cost %s seconds" % (f.__name__, time.time()-t))
            return res
        except Exception as error:
            print("function %s failed with error: %s" % (f.__name__, error))
    return execute

@measure_function
def my_function(numbers):
    odd_numbers = [n for n in numbers if n % 2 != 0]
    result = sum(odd_numbers)
    return result


print(my_function([1,2,3,5,6,7,8]))


@measure_function
def reverse_sentence(sentence):
    words = sentence.split()
    reversed_sentence = ' '.join(reversed(words))
    return reversed_sentence

print(reverse_sentence("Logs the execution time of each function"))

Answer 2

如果您想记录时间并捕获、报告和重新引发异常，那么您可以这样做：

from functools import wraps
from time import perf_counter
from pathlib import Path

ERROR_FILE = Path("except.txt")
TIMING_FILE = Path("timing.txt")


def my_wrapper(func):
    def report(path: Path, message: str | Exception):
        try:
            with path.open("a") as p:
                print(f"{func.__name__} {message}", file=p)
        except Exception:
            pass

    @wraps(func)
    def wrap_(*args, **kwargs):
        exception = None
        start = perf_counter()
        try:
            result = func(*args, **kwargs)
        except Exception as exception_:
            exception = exception_
            report(ERROR_FILE, exception)
        # if there was an exception in the wrapped function, the duration
        # will include the time taken to append to ERROR_FILE
        duration = perf_counter() - start
        report(TIMING_FILE, f"{duration=:.4f}")
        if exception is not None:
            raise exception
        return result

    return wrap_


@my_wrapper
def afunc():
    raise ValueError("That didn't work!")


@my_wrapper
def bfunc(x):
    print(x)


try:
    afunc()
except Exception as e:
    print(e)
bfunc(99)

输出到标准输出：

That didn't work!
99

except.txt 的最后一行

afunc That didn't work!

timing.txt 最后两行示例

afunc duration=0.0002
bfunc duration=0.0000