Python中工厂与组合模式联用的循环导入问题求解

循环依赖的核心原因：你将「子类导入注册」和「类查找实例化」两个逻辑耦合在同一个需要被双向引用的工厂模块中——工厂要导入子类完成注册，子类要导入工厂完成实例化，两个模块在顶层导入阶段互相等待对方完成初始化，必然触发循环导入错误。

不需要编写独立的双向依赖工厂模块，以下方案完全满足「子类独立存放、不修改基类代码、支持灵活扩展、无循环依赖」的需求，尤其适配Pipeline/Block组合模式场景。

方案1：基类内置自动注册机制（推荐，零额外依赖） #

利用Python3.6+原生支持的__init_subclass__钩子，把子类注册、实例化逻辑直接下沉到公共基类，从根源上消除双向导入的必要。

1. 首先在基类文件中实现注册和实例化逻辑，不需要提前导入任何子类：

# block_base.py
class Block:
    # 类级注册表，存储「类型标识 -> 子类」的映射关系
    _registry = {}

    def __init_subclass__(cls, block_type: str = None, **kwargs):
        # 子类被导入时自动触发，完成注册，不需要手动修改基类代码
        super().__init_subclass__(**kwargs)
        if block_type:
            cls._registry[block_type] = cls

    @classmethod
    def create(cls, block_config: dict) -> "Block":
        """统一实例化入口，所有子类直接继承该方法，不需要导入工厂"""
        # 从配置中取出类型标识
        b_type = block_config.pop("type")
        # 分离后继节点配置和当前块参数
        next_config = block_config.pop("next", None)
        # 实例化当前块
        instance = cls._registry[b_type](**block_config)
        # 递归实例化后继节点，自动兼容单个块/嵌套流水线
        instance.next = cls.create(next_config) if next_config else None
        return instance

    def __init__(self, **kwargs):
        # 预留自定义块的初始化逻辑
        self.next = None

2. 每个独立文件存放的子类，只需要导入基类，声明自己的类型标识即可，不需要导入工厂或其他子类：

# blocks/http_request.py
from block_base import Block

class HttpRequestBlock(Block, block_type="http_request"):
    def execute(self):
        print("执行HTTP请求逻辑")
        if self.next:
            self.next.execute()

# blocks/pipeline.py
from block_base import Block

class SubPipeline(Block, block_type="pipeline"):
    """嵌套流水线本身也作为Block的子类，完全兼容组合模式"""
    def execute(self):
        print("进入子流水线执行")
        if self.next:
            self.next.execute()

3. 只需要在项目入口处统一触发子类导入，完成注册即可：

# main.py 项目启动入口
from block_base import Block
# 导入所有需要用到的块，导入动作会自动触发注册
from blocks.http_request import HttpRequestBlock
from blocks.pipeline import SubPipeline
# 其他块同理

# 加载配置直接实例化整条流水线
import json
with open("pipeline_conf.json", "r") as f:
    conf = json.load(f)
pipeline = Block.create(conf)
pipeline.execute()

• 该方案的优势：
  • 完全无循环依赖：子类仅依赖基类，基类不依赖任何子类
  • 扩展零成本：新增块只需要新建独立文件写子类，不需要修改基类或工厂代码
  • 完美适配组合模式：嵌套流水线、单个块都遵循统一接口，节点不需要感知后继类型

如果不想每次新增块都在入口加导入代码，可以加个简单的自动扫描逻辑，启动时自动遍历块目录导入所有子类：

# 在Block类中新增自动发现方法
import importlib
import pathlib

@classmethod
def auto_load(cls, block_dir: str = "blocks"):
    for f in pathlib.Path(block_dir).glob("*.py"):
        if f.name.startswith("_"):
            continue
        importlib.import_module(f"{block_dir}.{f.stem}")

启动时调用Block.auto_load()即可，新块直接丢进blocks目录就会自动注册，不需要改任何入口代码。

方案2：延迟导入拆分工厂逻辑（兼容保留独立工厂模块的场景） #

如果你一定要保留独立的工厂模块，只需要把子类导入从模块顶层移到工厂函数内部，延迟到调用时再导入，就能避开顶层循环导入的问题：

# factory.py
def get_block_class(block_type: str):
    # 函数内部导入，调用时所有模块已经完成初始化，不会触发循环
    if block_type == "http_request":
        from blocks.http_request import HttpRequestBlock
        return HttpRequestBlock
    if block_type == "pipeline":
        from blocks.pipeline import SubPipeline
        return SubPipeline
    # 其他类型映射

• 该方案的缺点：每新增一个块都需要手动修改工厂函数的判断逻辑，维护成本比基类注册方案高。

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