面试题：Python装饰器在复杂业务逻辑编排场景中的应用

Question

在一个大型数据处理项目中，有一系列的数据处理函数，每个函数负责不同的处理步骤，如数据清洗、特征提取、模型训练等。这些函数需要按照特定顺序执行，并且每个步骤可能需要不同的前置条件判断。请设计一套基于Python装饰器的解决方案，能够灵活地编排这些函数的执行顺序，并在执行前进行前置条件检查，详细阐述设计思路并给出核心代码示例。

Answer 1

设计思路

1. 定义前置条件检查装饰器：创建一个装饰器，用于在函数执行前检查前置条件。这个装饰器接受一个检查函数作为参数，在被装饰函数执行前调用该检查函数。
2. 定义执行顺序管理装饰器：为了管理函数的执行顺序，我们可以使用一个带有状态的装饰器，在装饰器内部维护一个函数列表，用于记录函数执行顺序。
3. 实现数据传递：为了确保各个处理步骤之间的数据可以正确传递，每个函数的输入输出需要进行合理设计，并且在装饰器中进行相应的数据处理。

核心代码示例

# 前置条件检查装饰器
def precondition_check(condition_func):
    def decorator(func):
        def wrapper(*args, **kwargs):
            if not condition_func(*args, **kwargs):
                raise ValueError("前置条件不满足")
            return func(*args, **kwargs)
        return wrapper
    return decorator


# 执行顺序管理装饰器
class ExecutionOrderManager:
    def __init__(self):
        self.func_list = []

    def register(self, func):
        self.func_list.append(func)
        return func

    def execute(self, *args, **kwargs):
        data = args[0] if args else None
        for func in self.func_list:
            if data is not None:
                if isinstance(data, dict):
                    data = func(**data)
                else:
                    data = func(data)
            else:
                data = func()
        return data


# 示例前置条件检查函数
def data_exists(data):
    return data is not None


# 数据清洗函数
@precondition_check(data_exists)
def data_cleaning(data):
    print("执行数据清洗")
    return data + " 清洗后"


# 特征提取函数
@precondition_check(data_exists)
def feature_extraction(data):
    print("执行特征提取")
    return data + " 提取特征后"


# 模型训练函数
@precondition_check(data_exists)
def model_training(data):
    print("执行模型训练")
    return data + " 训练后"


# 使用执行顺序管理装饰器
execution_manager = ExecutionOrderManager()
data = "原始数据"

@execution_manager.register
def start():
    return data


@execution_manager.register
def step1():
    global data
    data = data_cleaning(data)
    return data


@execution_manager.register
def step2():
    global data
    data = feature_extraction(data)
    return data


@execution_manager.register
def step3():
    global data
    data = model_training(data)
    return data


result = execution_manager.execute()
print("最终结果:", result)

以上代码中：

1. precondition_check 装饰器实现了前置条件检查。
2. ExecutionOrderManager 类及其 register 和 execute 方法实现了函数执行顺序的管理。
3. 定义了示例的前置条件检查函数 data_exists 以及数据处理函数 data_cleaning、feature_extraction、model_training，并通过 ExecutionOrderManager 注册和执行这些函数。