Python重构时用元类实现DataSource动态请求方法是否合理

重构背景 #

当前承担老旧代码重构工作，原有作为Excel工作表实体的类直接内置数据库访问逻辑，计划拆分独立的DataSource类作为数据库访问中介层。
现有项目通过名为handlers的辅助函数接收参数、返回数据库查询结果，ExcelWorksheet类实现了异步执行handler的逻辑，每个子类会调用一个或多个handler，但原有实现逻辑耦合度高、可读性差，原有实现代码如下：

data_handler = {
    # 固定使用facebook渠道，提前注入参数
    "metrics": lambda *args: metrics_handler(*["facebook"] + list(args)),
    "top_content": lambda *args: content_handler(*["facebook"] + list(args)),
}

@property
def data_requirements(self):
    return [
        ["metrics", self.start, self.end, [self.mapping_id], [self.category_id], self.metrics],
        ["top_content", self.start, self.end, "total_actions", "user", [self.mapping_id], 10],
        ["top_content", self.start, self.end, "total_actions", "category", [self.category_id], 10],
        ["top_content", self.start, self.end, "comments_count", "user", [self.mapping_id], 5],
        ["top_content", self.start, self.end, "comments_count", "category", [self.category_id], 5],
        ["top_content", self.start, self.end, "shares_count", "user", [self.mapping_id], 5],
        ["top_content", self.start, self.end, "shares_count", "category", [self.category_id], 5],
    ]

@staticmethod
def _call_async(fns):
    """
    在线程池中异步调用传入的函数列表，返回结果生成器
    """
    futures = []
    with CloseConnectionThreadPoolExecutor(max_workers=4) as executor:
        for indx, f in enumerate(fns):
            LOGGER.warn("Getting data dependency %s out of %s" % (indx + 1, len(fns)))
            future = executor.submit(f)
            futures.append(future)

    return (i.result() for i in futures)

def require_data(self):
    """
    按定义顺序返回当前工作簿所有工作表需要的数据结果生成器
    调用self.data_handler获取对应key的处理函数：如果需求项是列表，第一个元素为key，后续元素为传入函数的参数

    数据需求格式示例：
    [
        'resource_1',
        'resource_2',
        ['resource_3', 'arg1', 'arg2'],
        ['resource_3', 'arg3', 'arg4']
    ]
    """
    if not self.data_handler:
        raise NotImplementedError("No data_handler specified")

    # 遍历所有数据需求，路由到对应handler
    data_fns = []
    for r in self.data_requirements:
        # 列表格式的需求：第一个元素是资源key，其余是调用参数
        if isinstance(r, list):
            key = r[0]
            args = r[1:]
            fn = self.data_handler[key](*args)
        else:
            fn = self.data_handler[r]

        data_fns.append(fn)

    return self._call_async(data_fns)

重构目标 #

拆分独立DataSource对象，优化handler调用逻辑、降低理解成本，期望的调用方式如下：

datasource.add_request_for_metrics(arg1,arg2,arg3...)
datasource.add_request_for_top_content(arg1,arg2,arg3...)
results = datasource.execute_requests()

由于项目中handler数量多、后续还会持续新增，手动为每个handler编写add_request_for_x方法维护成本很高，因此考虑使用Python元类，根据配置的handler映射自动生成对应的添加请求方法，初版元类实现代码如下：

class DataSourceMeta(type):
    """
    DataSource类的定义元类
    参数说明：
    data_handler_mappings - 数据类别与对应处理handler的映射字典，示例：
    {"metrics":metrics_handler, "time_series":time_series_handler}
    表示metrics类数据通过metrics_handler获取，time_series类数据通过time_series_handler获取

    元类会为字典中的每一项自动生成命名为`add_request_for_{数据类别名}`的实例方法，用于添加对应的数据请求
    所有添加的请求可以通过调用`execute_requests`方法执行，返回结果列表

    使用该元类的类需要实现以下接口：
    - add_request(): 将待执行的请求加入执行队列
    - _run_requests(): 定义请求的执行逻辑（例如同步/异步执行规则）
    - _clean_up(): 请求执行完成后的清理逻辑（例如清空请求队列）
    """
    def __new__(cls,*args,**kwargs):
        data_handler_mappings=kwargs.get("data_handler_mappings")
        newclass = super(DataSourceMeta,cls).__new__(cls,*args)
        for data_category,data_category_handler in data_handler_mappings.items():
            data_category=data_category.lower()
            method_name = "add_request_for_"+data_category
            def method(self,*args,**kwargs):
                newclass.add_request(self,data_category_handler(*args,**kwargs))
            setattr(newclass,method_name,method)
        def execute_requests(self):
            results = newclass._run_requests(self)
            newclass._clean_up(self)
            return results
        newclass.execute_requests = execute_requests
        return newclass

方案评估与优化建议 #

方案合理性 #

你拆分独立DataSource层的思路完全正确——把数据访问逻辑从Excel工作表实体类中剥离，符合单一职责原则，后续调整数据库查询逻辑不会影响工作表本身的业务计算逻辑，从根源上解决旧代码耦合度高的问题。
用自动生成方法替代重复手写add_request_for_x的思路也没问题，handler多、迭代频繁的场景下，自动生成能彻底避免手动编码带来的命名不一致、参数传错等低级错误，长期维护成本低。
元类的核心方向是对的：读取handler映射配置、动态挂载对应方法、统一封装请求执行的入口逻辑，比旧代码里硬编码数据需求列表的可读性高很多。

现存问题和可优化点 #

• 修复闭包延迟绑定bug
  现在循环里定义的method存在Python闭包的经典问题：所有动态生成的方法，最终都会引用循环最后一次迭代的data_category_handler，不管调用哪个add_request_for_xx方法，实际执行的都是最后一个handler，逻辑完全错误。
  修复方式很简单，把当前循环的handler作为函数默认参数传进去，在函数定义阶段就把值绑定死：

  def method(self, *args, _handler=data_category_handler, **kwargs):
      self.add_request(_handler(*args, **kwargs))
  

• 修正类创建的参数传递逻辑
  元类__new__里接收的kwargs是类定义时传入的自定义参数，调用父类__new__创建类的时候，要先把自定义的data_handler_mappings从kwargs里pop出去，不然Python会报意外关键字参数的错误。另外调用实例方法的时候不需要硬传self，直接通过实例调用即可，比如写self.add_request(...)，不要写newclass.add_request(self, ...)。
• 补全边界校验
  现在如果有人用这个元类的时候忘了传data_handler_mappings配置，循环遍历的时候直接会抛NoneType错误，提示信息完全没法定位问题。建议加两层校验：一是如果没传handler映射，要么抛明确的初始化异常，要么给空字典作为默认值；二是在类创建阶段就检查使用元类的类有没有实现add_request、_run_requests、_clean_up三个必须的接口方法，缺哪个直接抛异常，不要等到实际调用方法的时候才报错。
• 补全方法元信息，提升开发体验
  现在动态生成的方法没有文档字符串、没有参数签名，IDE没法做自动补全，调试的时候也看不到方法对应的上下文。可以用functools.wraps把原handler的元信息绑定到动态生成的方法上，开发体验和手写方法完全一致。
• 做好旧逻辑兼容
  旧代码里的metrics、top_content两个handler都要固定传"facebook"作为第一个参数，配置data_handler_mappings的时候可以用functools.partial提前把这个固定参数绑定好，业务层调用add_request_for_xxx的时候不用重复传这个固定值，和旧逻辑完全兼容，不用改上层调用代码。
• 元类替代方案可选
  如果团队其他开发对元类熟悉度不高，其实不用元类也能实现同样的效果：重写__getattr__方法，当访问的属性名以add_request_for_开头的时候，动态解析出对应的handler，返回包装好的请求添加函数即可，逻辑更直白，调试成本更低。当然如果团队平时就有用元类的习惯，保留元类方案完全没问题。

内容的提问来源于stack exchange，提问作者AlanKalane