动态生成Pydantic模型作为函数类型注解引发mypy报错的解决问询

问题背景 #

采用插件式架构，启动时将Pydantic模型注册到中央注册表，动态添加为现有模型的Union字段，该类用于Dagster框架的UI YAML配置。Dagster依赖函数的config参数类型注解来注册配置类，但通过工厂函数创建asset时，mypy报错：

'Variable "assets.config_class" is not valid as a type [valid-type]'

目前通过# type: ignore忽略错误，需寻求同时满足框架需求与mypy检查的方案。

核心解决方案 #

1. 用泛型+TypeVar实现类型安全的工厂函数 #

mypy不允许直接使用变量作为类型注解，通过定义绑定到BaseModel的TypeVar，让工厂函数能被静态检查识别：

from typing import TypeVar, Type, List
from pydantic import BaseModel, create_model
from dagster import asset, AssetsDefinition

# 定义绑定到BaseModel的类型变量
T = TypeVar('T', bound=BaseModel)

def create_asset(identifier: str, config_class: Type[T]) -> AssetsDefinition:
    @asset(key=identifier)
    def _inner(config: T) -> None:
        # 业务逻辑处理
        pass
    return _inner

2. 动态生成带Union字段的配置模型 #

替换原MyBase中的Any类型，用create_model动态生成包含所有插件配置类的Union字段，同时让mypy能识别其类型：

# 注册所有插件配置类
plugin_configs: List[Type[BaseModel]] = register_plugins()

# 动态生成包含Union字段的配置模型
# 用tuple打包插件类生成Union类型
config_union = tuple(plugin_configs)
config_class = create_model(
    "MyBaseWithPlugins",
    __base__=BaseModel,
    configs=(config_union, ...)  # 替换原Any类型为Union[所有插件配置类]
)

3. 保留基础模型字段的兼容方案 #

如果需要保留原MyBase的基础字段，可将其作为基类传入create_model：

class MyBase(BaseModel):
    common_field: str = "default"

config_class = create_model(
    "MyBaseWithPlugins",
    __base__=MyBase,
    configs=(config_union, ...)
)

4. 验证mypy检查 #

修改后的代码无需# type: ignore，mypy能正确识别config参数的类型为动态生成的配置模型，同时满足Dagster对config参数类型注解的要求。

完整示例代码 #

from typing import TypeVar, Type, List
from pydantic import BaseModel, create_model
from dagster import asset, AssetsDefinition

# 类型变量定义
T = TypeVar('T', bound=BaseModel)

def create_asset(identifier: str, config_class: Type[T]) -> AssetsDefinition:
    @asset(key=identifier)
    def _inner(config: T) -> None:
        print(f"处理配置: {config.dict()}")
    return _inner

# 模拟插件注册函数
def register_plugins() -> List[Type[BaseModel]]:
    class PluginConfig1(BaseModel):
        plugin1_param: int

    class PluginConfig2(BaseModel):
        plugin2_param: str

    return [PluginConfig1, PluginConfig2]

# 基础配置模型
class MyBase(BaseModel):
    common_config: str = "common"

# 动态生成带插件配置的模型
plugin_configs = register_plugins()
config_union = tuple(plugin_configs)
config_class = create_model(
    "MyBaseWithPlugins",
    __base__=MyBase,
    configs=(config_union, ...)
)

# 加载资产ID并创建资产
def load_asset_ids() -> List[str]:
    return ["asset_1", "asset_2"]

my_list_of_ids = load_asset_ids()
assets = [create_asset(id, config_class) for id in my_list_of_ids]

内容的提问来源于stack exchange，提问作者Martin O Leary