Python3.x链式生成器与过滤器框架的文件组织优化问询

核心问题拆解 #

• 父类Renderer硬编码子类方法，耦合度极高，扩展困难
• 模块静态导入导致循环依赖（父类导入子类，子类继承父类）

重构方案1：基于组合的动态链式容器 #

放弃让生成器/过滤器继承Renderer，改用组合模式实现链式调用核心容器，通过__getattr__动态加载组件，彻底解耦核心逻辑与业务组件。

代码实现 #

核心容器类（core/chain.py）

from typing import Any, Callable
import importlib

class Chain:
    def __init__(self, current_data: Any = None):
        self.current_data = current_data

    def __getattr__(self, name: str) -> Callable:
        # 优先查找过滤器组件
        try:
            filter_mod = importlib.import_module(f".filters.{name}", package="your_framework.core")
            filter_cls = getattr(filter_mod, name.capitalize())
            
            def wrapper(*args, **kwargs):
                self.current_data = filter_cls(*args, **kwargs).process(self.current_data)
                return self
            return wrapper
        except (ImportError, AttributeError):
            # 查找生成器组件
            try:
                gen_mod = importlib.import_module(f".generators.{name}", package="your_framework.core")
                gen_cls = getattr(gen_mod, name.capitalize())
                
                def wrapper(*args, **kwargs):
                    self.current_data = gen_cls(*args, **kwargs).generate()
                    return self
                return wrapper
            except (ImportError, AttributeError):
                raise AttributeError(f"No component named '{name}' found")

    def get_result(self) -> Any:
        return self.current_data

过滤器组件示例（core/filters/filt1.py）

class Filt1:
    def __init__(self, threshold: int):
        self.threshold = threshold

    def process(self, data: list[int]) -> list[int]:
        return [x for x in data if x > self.threshold]

生成器组件示例（core/generators/gen1.py）

class Gen1:
    def generate(self) -> list[int]:
        return [1, 2, 3, 4, 5, 6]

对外导出（your_framework/__init__.py）

from .core.chain import Chain as rm

使用方式 #

result = rm().gen1().filt1(3).get_result()
# 输出: [4,5,6]

重构方案2：元类自动注册机制 #

通过元类在组件定义时自动注册到核心类，无需手动维护方法映射，同时避免静态导入导致的循环依赖。

代码实现 #

基类与元类（core/base.py）

class ComponentMeta(type):
    """元类：自动注册生成器/过滤器组件"""
    registry = {}

    def __new__(cls, name, bases, attrs):
        new_cls = super().__new__(cls, name, bases, attrs)
        # 以类名小写作为链式调用方法名
        ComponentMeta.registry[name.lower()] = new_cls
        return new_cls

class Renderer:
    def __init__(self, current_data: Any = None):
        self.current_data = current_data

    def __getattr__(self, name: str) -> Callable:
        if name not in ComponentMeta.registry:
            raise AttributeError(f"No component named '{name}'")
        
        component_cls = ComponentMeta.registry[name]
        def wrapper(*args, **kwargs):
            comp = component_cls(*args, **kwargs)
            if hasattr(comp, "generate"):
                self.current_data = comp.generate()
            elif hasattr(comp, "process"):
                self.current_data = comp.process(self.current_data)
            return self
        return wrapper

    def get_result(self) -> Any:
        return self.current_data

生成器组件（core/generators/gen1.py）

from ..base import ComponentMeta

class Gen1(metaclass=ComponentMeta):
    def generate(self) -> list[int]:
        return [1,2,3,4,5]

过滤器组件（core/filters/filt1.py）

from ..base import ComponentMeta

class Filt1(metaclass=ComponentMeta):
    def __init__(self, val: int):
        self.val = val
    def process(self, data: list[int]) -> list[int]:
        return [x for x in data if x > self.val]

使用方式 #

result = rm.Renderer().gen1().filt1(3).get_result()

优雅的文件组织建议 #

your_framework/
├── __init__.py          # 对外导出核心类（如rm）
├── core/
│   ├── __init__.py
│   ├── chain.py         # 方案1的链式容器
│   ├── base.py          # 方案2的元类与基类
│   ├── generators/
│   │   ├── __init__.py  # 可选：批量导入生成器组件
│   │   ├── gen1.py
│   │   └── gen2.py
│   └── filters/
│       ├── __init__.py  # 可选：批量导入过滤器组件
│       ├── filt1.py
│       └── filt2.py

关键注意事项 #

• 避免在核心模块中静态导入组件，改用动态导入或元类注册，彻底解决循环依赖
• 所有组件遵循统一接口：生成器实现generate()，过滤器实现process(data)，便于核心容器统一处理
• 模块__init__.py仅导出必要的公共API，避免暴露内部细节

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