装饰器到底是什么语法糖？——Python函数式编程

# 装饰器的等价写法
 
# 用@语法糖
@timer
def process_data():
    pass
 
# 等价于
def process_data():
    pass
process_data = timer(process_data)  # timer返回一个新函数
 
# timer装饰器的实现
import time
def timer(func):
    def wrapper(*args, **kwargs):
        start = time.time()
        result = func(*args, **kwargs)  # 调用原函数
        elapsed = time.time() - start
        print(f'{func.__name__} took {elapsed:.2f}s')
        return result
    return wrapper  # 返回包装后的新函数
 
# 闭包：wrapper函数引用了外层timer函数的func变量
# 即使timer函数已经返回，wrapper仍然能访问func

# 带参数的装饰器需要多一层函数嵌套
 
def retry(max_attempts=3, delay=1):
    def decorator(func):              # 第二层：接收函数
        def wrapper(*args, **kwargs):  # 第三层：实际执行
            for attempt in range(max_attempts):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if attempt < max_attempts - 1:
                        print(f'Attempt {attempt+1} failed: {e}')
                        time.sleep(delay)
                    else:
                        raise  # 最后一次仍失败则抛出
        return wrapper
    return decorator  # 返回装饰器函数
 
# 使用：
@retry(max_attempts=5, delay=2)
def fetch_data_from_api(url):
    return requests.get(url).json()
 
# 等价于：fetch_data_from_api = retry(5, 2)(fetch_data_from_api)

实战经验：ETL脚本的@retry装饰器

在某短剧出海公司，ETL脚本经常需要调用外部API（广告平台获取投放数据）。网络抖动导致API偶尔超时，脚本直接失败。我写了一个通用的@retry装饰器：

import functools, time, logging
 
def retry(max_attempts=3, delay=5, exceptions=(Exception,)):
    def decorator(func):
        @functools.wraps(func)  # 保留原函数的名字和文档
        def wrapper(*args, **kwargs):
            for attempt in range(1, max_attempts + 1):
                try:
                    return func(*args, **kwargs)
                except exceptions as e:
                    logging.warning(
                        f'{func.__name__} attempt {attempt}/{max_attempts} '
                        f'failed: {e}')
                    if attempt < max_attempts:
                        time.sleep(delay * attempt)  # 递增等待
                    else:
                        logging.error(f'{func.__name__} all attempts failed')
                        raise
        return wrapper
    return decorator
 
# 使用：所有API调用加上@retry就有了重试能力
@retry(max_attempts=5, delay=3, exceptions=(TimeoutError, ConnectionError))
def fetch_ad_data(date, platform):
    return requests.get(f'{API_URL}?date={date}', timeout=30).json()

效果：加了@retry后API调用的成功率从95%提升到99.9%——大部分超时只是偶发的网络抖动，重试一次就成功了。而且代码非常干净——业务逻辑和重试逻辑完全分离。

functools.wraps——被忽视的重要细节

# 不加@functools.wraps的问题
def timer(func):
    def wrapper(*args, **kwargs):
        pass
    return wrapper
 
@timer
def process_data():
    '''处理数据的函数'''
    pass
 
print(process_data.__name__)  # 'wrapper'  不是'process_data'!
print(process_data.__doc__)   # None  文档字符串丢了！
 
# 加了@functools.wraps(func)后
def timer(func):
    @functools.wraps(func)  # 复制原函数的元信息
    def wrapper(*args, **kwargs):
        pass
    return wrapper
 
print(process_data.__name__)  # 'process_data'  正确！
print(process_data.__doc__)   # '处理数据的函数'  正确！
 
# 在Airflow中这特别重要——Airflow通过函数名识别Task
# 不加wraps会导致Task名字变成'wrapper'

lru_cache——内置的缓存装饰器

from functools import lru_cache
 
# 缓存函数结果，相同参数不重复计算
@lru_cache(maxsize=1024)
def get_country_name(country_code):
    # 假设这是一个耗时的查询
    return db.query(f"SELECT name FROM countries WHERE code='{country_code}'")
 
# 第一次调用 get_country_name('US')：查数据库，缓存结果
# 第二次调用 get_country_name('US')：直接返回缓存，不查数据库
 
# 大数据场景：维度表查询缓存
# 处理100万行数据时，country_code只有50种
# 没有缓存：100万次数据库查询
# 有lru_cache：50次查询 + 999950次缓存命中
 
# 注意：lru_cache要求参数是可哈希的（不能传list/dict）
# maxsize=None表示无限缓存（小心内存！）

面试高频追问

追问一：装饰器可以装饰类吗？

可以。类装饰器接收一个类作为参数返回一个修改后的类。常见用途：@dataclass就是一个类装饰器——它自动给类添加__init__、__repr__、__eq__等方法。你也可以写自定义类装饰器给所有方法加日志。

追问二：多个装饰器的执行顺序是什么？

从下往上装饰，从上往下执行。@A @B @C def f(): 等价于 f = A(B(C(f)))。所以C先装饰f，然后B装饰C(f)的结果，最后A装饰B(C(f))的结果。执行时A的wrapper先执行，然后B的，最后C的，最后才是f本身。

追问三：闭包是什么？和装饰器有什么关系？

闭包是引用了外层函数局部变量的内层函数。装饰器的wrapper函数就是一个闭包——它引用了外层decorator函数的func参数。闭包的特点是外层函数返回后，内层函数仍然能访问外层的变量。这是Python实现装饰器的底层机制。

总结

装饰器面试讲清楚三层：① 本质（高阶函数+闭包，@decorator等价于func=decorator(func)）；② 带参数的装饰器（三层嵌套——参数层/函数层/执行层）；③ 大数据场景应用（@retry重试、@udf注册Spark函数、@lru_cache维度缓存、@functools.wraps保留元信息）。能讲出ETL脚本@retry装饰器的实际应用和API成功率从95%提升到99.9%的效果，面试官会觉得你能把Python特性用在实际工程中。

下一篇聊生成器和迭代器：yield和return有什么区别？大数据流式处理为什么需要生成器？

你们写过什么实用的装饰器？评论区聊聊～

关注「Wood聊大数据」| 9年大数据老兵，只讲大数据面试干货