装饰器提供了一种强大且Python风格的方式来修改或加强函数和方法。它们允许您在现有代码周围添加额外功能,而无需永久改变原始函数的定义。这有助于提升代码重用性和关注点分离,这在构建数据处理管道或机器学习工作流时是非常有益的做法。本质上,装饰器是一个可调用对象(通常是函数),它接受另一个函数作为输入并返回一个新函数。直接放置在函数定义上方的 @decorator_name 语法是一种语法糖,简化了此过程。来看一下这个基本结构:import functools def my_decorator(func): @functools.wraps(func) # 保留原始函数元数据 def wrapper(*args, **kwargs): # 在原始函数运行之前执行的代码 print(f"在调用 {func.__name__} 之前发生了一些事情。") result = func(*args, **kwargs) # 调用原始函数 # 在原始函数运行之后执行的代码 print(f"在 {func.__name__} 完成之后发生了一些事情。") return result return wrapper @my_decorator def say_hello(name): """问候用户。""" print(f"你好, {name}!") # 调用被装饰函数 say_hello("Data Scientist") # 输出: # 在调用 say_hello 之前发生了一些事情。 # 你好, Data Scientist! # 在 say_hello 完成之后发生了一些事情。在这里,my_decorator 是装饰器函数。它定义了一个内部函数 wrapper,其中包含额外逻辑。wrapper 函数会调用传递给装饰器的原始函数 (func)。say_hello 上方的 @my_decorator 语法等同于在函数定义之后编写 say_hello = my_decorator(say_hello)。注意在装饰器内部使用 @functools.wraps(func)。这是一个辅助装饰器,它通过复制 __name__、__doc__ 和参数签名等属性,使 wrapper 函数看起来像原始函数 (func)。如果没有 @functools.wraps,内省工具(以及其他潜在代码)将看到 wrapper 函数的信息,而不是 say_hello 函数的信息。装饰器工作原理图解您可以将装饰器看作是在原始函数周围应用了一个层:digraph G { rankdir=TB; node [shape=box, style=rounded, fontname="Helvetica", fontsize=10]; edge [fontsize=10]; subgraph cluster_decorator { label = "my_decorator"; bgcolor="#e9ecef"; style=filled; node [fillcolor="#ffffff", style=filled]; wrapper [label="wrapper(*args, **kwargs)"]; before_code [label="print(...) # 之前", shape=plaintext]; after_code [label="print(...) # 之后", shape=plaintext]; original_call [label="result = func(*args, **kwargs)"]; before_code -> original_call; original_call -> after_code; wrapper -> before_code [style=invis]; # 布局提示 } original_func [label="say_hello(name)", shape=box, style="rounded,filled", fillcolor="#a5d8ff"]; call_decorated [label="调用 say_hello(...)"]; call_decorated -> wrapper [label="执行 wrapper"]; original_call -> original_func [label="调用原始函数"]; }装饰器 (my_decorator) 定义了一个 wrapper。当被装饰函数 (say_hello) 被调用时,wrapper 执行,在调用原始函数 (func) 之前和之后运行代码。数据科学中的常见应用场景装饰器对于添加与数据分析和机器学习任务相关的跨领域功能特别有用:函数执行计时: 衡量特定数据处理步骤或计算的耗时对于优化很重要。import time import functools import pandas as pd import numpy as np def timer(func): @functools.wraps(func) def wrapper_timer(*args, **kwargs): start_time = time.perf_counter() # 比 time.time() 更精确 value = func(*args, **kwargs) end_time = time.perf_counter() run_time = end_time - start_time print(f"' {func.__name__!r} ' 完成于 {run_time:.4f} 秒") return value return wrapper_timer @timer def simulate_data_processing(rows=1000000): """模拟一个可能耗时的数据操作。""" df = pd.DataFrame(np.random.rand(rows, 5), columns=list('ABCDE')) # 模拟一些计算 result = df['A'] * np.sin(df['B']) - df['C'] * np.cos(df['D']) time.sleep(0.5) # 模拟 I/O 或其他延迟 return result.mean() mean_value = simulate_data_processing(rows=500000) print(f"平均结果: {mean_value}") # 示例输出: # 'simulate_data_processing' 完成于 0.6123 秒 # 平均结果: 0.001...日志记录: 追踪函数调用、参数或结果对于调试复杂管道非常有帮助。import logging import functools logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') def logger(func): @functools.wraps(func) def wrapper_logger(*args, **kwargs): logging.info(f"调用 {func.__name__},参数: {args},关键字参数: {kwargs}") try: result = func(*args, **kwargs) logging.info(f"{func.__name__} 返回: {type(result)}") return result except Exception as e: logging.error(f" {func.__name__} 中发生异常: {e}", exc_info=True) raise # 记录后重新引发异常 return wrapper_logger @logger def load_data(filepath): """加载数据,可能引发错误。""" if not filepath.endswith(".csv"): raise ValueError("文件类型无效,预期为 .csv") # 模拟加载数据 print(f"从 {filepath} 加载数据...") return pd.DataFrame({'col1': [1, 2], 'col2': [3, 4]}) # 虚拟 DataFrame try: df = load_data("my_data.csv") # df_error = load_data("my_data.txt") # 取消注释以查看错误日志 except ValueError as e: print(f"捕获到预期错误: {e}") # 示例日志输出: # 2023-10-27 10:30:00,123 - INFO - 调用 load_data,参数: ('my_data.csv',),关键字参数: {} # 从 my_data.csv 加载数据... # 2023-10-27 10:30:00,124 - INFO - load_data 返回: <class 'pandas.core.frame.DataFrame'> # (如果错误情况取消注释) # 2023-10-27 10:30:00,125 - INFO - 调用 load_data,参数: ('my_data.txt',),关键字参数: {} # 2023-10-27 10:30:00,126 - ERROR - load_data 中发生异常: 文件类型无效,预期为 .csv # Traceback (most recent call last): ... # 捕获到预期错误: 文件类型无效,预期为 .csv输入验证: 在继续之前,确保函数接收的数据格式正确(例如,具有特定列的 DataFrame)。import functools import pandas as pd def requires_columns(required_cols): def decorator(func): @functools.wraps(func) def wrapper_validator(*args, **kwargs): # 假设 DataFrame 是第一个位置参数 if args and isinstance(args[0], pd.DataFrame): df = args[0] missing_cols = set(required_cols) - set(df.columns) if missing_cols: raise ValueError(f" {func.__name__} 的 DataFrame 中缺少所需列: {missing_cols}") else: # 可以为 kwargs 或其他位置添加更复杂的检查 pass # 或者如果未在预期位置找到 DF 则引发错误 return func(*args, **kwargs) return wrapper_validator return decorator @requires_columns(['feature1', 'target']) def process_features(df): """处理 DataFrame 中的特定特征。""" print("正在处理特征...") # 实际处理逻辑在此处 return df['feature1'] * 2 data_ok = pd.DataFrame({'feature1': [1, 2, 3], 'target': [0, 1, 0], 'extra': [5, 6, 7]}) data_bad = pd.DataFrame({'feature_typo': [1, 2, 3], 'target': [0, 1, 0]}) result = process_features(data_ok) # 正常运行 try: process_features(data_bad) # 引发 ValueError except ValueError as e: print(f"验证失败: {e}") # 输出: # 正在处理特征... # 验证失败: process_features 的 DataFrame 中缺少所需列: {'feature1'}此示例还展示了带参数的装饰器。requires_columns 是一个工厂函数,它接受所需列列表并返回实际的装饰器函数。这允许定制装饰器的行为。记忆化(缓存): 存储计算开销大的函数调用的结果,并在相同输入再次出现时返回缓存结果。Python 的 functools 模块为此提供了 lru_cache(最近最少使用缓存)。import functools import time @functools.lru_cache(maxsize=None) # None 表示缓存大小无限制 def expensive_calculation(a, b): """模拟一个开销大的计算。""" print(f"正在对 ({a}, {b}) 执行开销大的计算...") time.sleep(1) # 模拟工作 return a + b * b print(expensive_calculation(2, 3)) # 运行计算 print(expensive_calculation(5, 2)) # 运行计算 print(expensive_calculation(2, 3)) # 立即返回缓存结果 print(expensive_calculation(5, 2)) # 立即返回缓存结果 # 输出: # 正在对 (2, 3) 执行开销大的计算... # 11 # 正在对 (5, 2) 执行开销大的计算... # 9 # 11 # 9尽管 NumPy 和 Pandas 操作通常在内部高度优化,但 lru_cache 对于您的工作流中重复对相同输入执行大量计算的自定义 Python 函数很有益处。装饰器堆叠您可以将多个装饰器应用于单个函数。它们在语法上从下到上应用,但执行时从上到下(最外层包装器首先运行)。@timer @logger # @requires_columns(['input']) # 示例:添加验证 def complex_step(data): # ... 处理逻辑 ... print("正在执行复杂步骤...") time.sleep(0.2) return "Done" complex_step("Some Input Data") # 日志输出将显示计时器在日志消息前后开始/结束。 # 执行顺序:计时器包装器 -> 日志记录器包装器 -> 原始 complex_step装饰器是一种灵活的工具,可以在不使核心逻辑变得混乱的情况下,为您的函数添加日志记录、计时、验证或缓存等行为。熟练使用它们能帮助你编写更模块化、可重用和易于维护的 Python 代码,这在工作流可能变得复杂的数据科学和机器学习项目中非常有优势。