装饰器是Python的语言特性，它可以无侵入地修改、增强、包装函数或者类的行为。

常见的装饰器用途有：日志、计时器、缓存、权限控制、函数重试、限制调用频率、打印调用栈等。下面列出常用的装饰器代码。

重试

在抓取数据时常用的功能，控制函数在失败后自动进行重试，并限制最大重试次数和间隔时间。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
import time
from functools import wraps

def retry(max_tries=3, delay_seconds=1):
    def decorator_retry(func):
        @wraps(func)
        def wrapper_retry(*args, **kwargs):
            tries = 0
            while tries < max_tries:
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    tries += 1
                    if tries == max_tries:
                        raise e
                    time.sleep(delay_seconds)
        return wrapper_retry
    return decorator_retry

@retry(max_tries=5, delay_seconds=2)
def call_dummy_api(url):
    response = requests.get(url)
    return response

functools.wraps用来避免调用包装函数时丢失原始函数的元信息，例如函数名称、参数列表和文档字符串。它创建一个包装函数，该函数将保留原始函数的名称、文档字符串、参数列表和注释。它的作用是将被装饰函数的元信息复制到包装函数中，以便在调用包装函数时，可以正确地显示原始函数的信息。

缓存

用装饰器缓存计算过程，减少重复计算。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
def memoize(func):
    cache = {}
    def wrapper(*args):
        if args in cache:
            return cache[args]
        else:
            result = func(*args)
            cache[args] = result
            return result
    return wrapper

@memoize
def fibonacci(n):
    if n <= 1:
        return n
    else:
        return fibonacci(n-1) + fibonacci(n-2)

计时器

记录函数运行时长。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
import time

def timing_decorator(func):
    def wrapper(*args, **kwargs):
        start_time = time.time()
        result = func(*args, **kwargs)
        end_time = time.time()
        print(f"Function {func.__name__} took {end_time - start_time} seconds to run.")
        return result
    return wrapper

@timing_decorator
def my_function():
    # some code here
    time.sleep(1)  # simulate some time-consuming operation
    return

日志

通过装饰器记录函数调用记录

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
import logging
import functools

logging.basicConfig(level=logging.INFO)

def log_execution(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        logging.info(f"Executing {func.__name__}")
        result = func(*args, **kwargs)
        logging.info(f"Finished executing {func.__name__}")
        return result
    return wrapper

@log_execution
def extract_data(source):
    # extract data from source
    data = ...

    return data

@log_execution
def transform_data(data):
    # transform data
    transformed_data = ...

    return transformed_data

@log_execution
def load_data(data, target):
    # load data into target
    ...

def main():
    # extract data
    data = extract_data(source)

    # transform data
    transformed_data = transform_data(data)

    # load data
    load_data(transformed_data, target)

结果：

1
2
3
4
5
6
INFO:root:Executing extract_data
INFO:root:Finished executing extract_data
INFO:root:Executing transform_data
INFO:root:Finished executing transform_data
INFO:root:Executing load_data
INFO:root:Finished executing load_data

另外还可以用装饰器实现通知功能，发生异常时发送邮件或消息。