趋近智
装饰器提供了一种强大且Python风格的方式来修改或加强函数和方法。它们允许您在现有代码周围添加额外功能,而无需永久改变原始函数的定义。这有助于提升代码重用性和关注点分离,这在构建数据处理管道或机器学习 (machine learning)工作流时是非常有益的做法。
本质上,装饰器是一个可调用对象(通常是函数),它接受另一个函数作为输入并返回一个新函数。直接放置在函数定义上方的 @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__ 和参数 (parameter)签名等属性,使 wrapper 函数看起来像原始函数 (func)。如果没有 @functools.wraps,内省工具(以及其他潜在代码)将看到 wrapper 函数的信息,而不是 say_hello 函数的信息。
您可以将装饰器看作是在原始函数周围应用了一个层:
装饰器 (
my_decorator) 定义了一个wrapper。当被装饰函数 (say_hello) 被调用时,wrapper执行,在调用原始函数 (func) 之前和之后运行代码。
装饰器对于添加与数据分析和机器学习 (machine learning)任务相关的跨领域功能特别有用:
函数执行计时: 衡量特定数据处理步骤或计算的耗时对于优化很重要。
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...
日志记录: 追踪函数调用、参数 (parameter)或结果对于调试复杂管道非常有帮助。
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 代码,这在工作流可能变得复杂的数据科学和机器学习 (machine learning)项目中非常有优势。
这部分内容有帮助吗?
© 2026 ApX Machine LearningAI伦理与透明度•