如何以装饰器为标签动态导入工作流所需类？

你的思路完全正确，这种方式和Flask通过装饰器注册路由的逻辑一致——我们可以让装饰器把目标类自动收集到全局注册表，工作流管理器直接从注册表提取需要的类，无需全量导入后再做类型检查。

具体实现步骤： #

1. 实现装饰器与注册表

先创建一个专门的装饰器模块，用来定义标记装饰器和存放类的注册表：

# module_decorators.py
# 按模块类型分类的注册表
MODULE_REGISTRY = {
    'prediction': [],
    'crawler': []
}

def prediction_module(cls):
    # 将被装饰的类添加到预测模块注册表
    MODULE_REGISTRY['prediction'].append(cls)
    return cls  # 返回原类，不影响类的正常使用

def crawler_module(cls):
    # 将被装饰的类添加到爬虫模块注册表
    MODULE_REGISTRY['crawler'].append(cls)
    return cls

2. 用户自定义模块类（classes.py）

用户在编写自定义类时，直接导入装饰器标记目标类即可：

# classes.py
from module_decorators import prediction_module, crawler_module

@prediction_module
class PredictionModule:
    def run(self):
        print("执行预测模块逻辑")

@crawler_module
class CrawlerModuleWithCustomName:
    def run(self):
        print("执行自定义爬虫模块逻辑")

# 无关类不受影响，不会被收集
@some_other_decorator
class OtherIrrelevantClassUsedSomewhereElse:
    pass

3. 工作流管理器获取目标类

工作流管理器只需导入classes.py触发装饰器的注册逻辑，然后直接从注册表提取需要的类：

# workflow_manager.py
from module_decorators import MODULE_REGISTRY
# 导入classes.py，触发装饰器执行，完成类的注册（无需显式引用类）
import classes

# 获取所有标记为预测模块的类
prediction_modules = MODULE_REGISTRY['prediction']
# 获取所有标记为爬虫模块的类
crawler_modules = MODULE_REGISTRY['crawler']

# 实例化并运行模块
for module_cls in prediction_modules:
    module_instance = module_cls()
    module_instance.run()

for module_cls in crawler_modules:
    module_instance = module_cls()
    module_instance.run()

扩展说明： #

• 如需支持更多模块类型，只需在MODULE_REGISTRY中新增对应键，并添加同名装饰器即可
• 被装饰的类依然可以正常被其他模块导入使用，装饰器仅做注册标记
• 这种方式完全避免了from classes import *后逐个检查类类型的冗余操作，逻辑更清晰高效

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