FastAPI应用中动态创建Enum并在SQLAlchemy模型中使用的无循环依赖方案

我明白你的困境：配置在main.py里加载，生成SchemaNames需要配置里的prefix，但SQLAlchemy模型要用到SchemaNames，同时API模块又导入了模型，导致循环依赖，而且你不想用全局变量来绕开这个问题。下面给你几个符合最佳实践的解决方案，按推荐程度排序：

方案一：使用模型工厂函数（最推荐） #

这个方案的核心是把模型定义成工厂函数，显式接收初始化好的SchemaNames作为参数，完全解耦配置、Enum和模型之间的依赖，从根源上避免循环依赖。

步骤1：保持enums.py不变 #

你的create_schema_names函数已经写得很好了，继续保留：

from enum import Enum, Type

def create_schema_names(prefix: str) -> Type[Enum]:
    class SchemaNames(str, Enum):
        RAW = prefix + "_RAW"
        STAGING = prefix + "_STAGING"
        TRANSFORMED = prefix + "_TRANSFORMED"
    return SchemaNames

步骤2：修改my_model.py为工厂函数 #

不再直接定义模型类，而是写一个接收SchemaNames的工厂，动态生成模型：

from sqlalchemy import Column, String, Integer
from sqlalchemy.ext.declarative import declarative_base
from enum import Enum

Base = declarative_base()

def create_models(SchemaNames: Enum):
    """根据传入的SchemaNames动态生成SQLAlchemy模型"""
    class RawData(Base):
        __tablename__ = "raw_data"
        __table_args__ = {"schema": SchemaNames.RAW.value}
        id = Column(Integer, primary_key=True)
        data = Column(String)

    class StagingData(Base):
        __tablename__ = "staging_data"
        __table_args__ = {"schema": SchemaNames.STAGING.value}
        id = Column(Integer, primary_key=True)
        processed_data = Column(String)

    return RawData, StagingData

步骤3：调整API注册逻辑，接收模型参数 #

修改register_api_2，让它不再直接导入模型，而是接收工厂生成的模型实例：

from fastapi import APIRouter

router = APIRouter()

def register_api_2(app, raw_model, staging_model):
    """接收动态生成的模型，注册API路由"""
    @router.get("/raw-data")
    async def get_raw_data():
        # 这里可以直接使用raw_model进行数据库操作
        return {"message": "Raw data endpoint"}

    @router.get("/staging-data")
    async def get_staging_data():
        # 使用staging_model
        return {"message": "Staging data endpoint"}

    app.include_router(router)

步骤4：在main.py中按顺序初始化依赖 #

现在可以按“加载配置 → 创建Enum → 生成模型 → 注册API”的顺序执行，完全没有循环：

from fastapi import FastAPI
from app.api import register_api_1
from app.utils.enums import create_schema_names
from app.models.my_model import create_models
import uvicorn

def make_config():
    # 模拟从YAML解析配置的逻辑
    class ConfigClass:
        prefix = "MY_APP"
        webserver = type('obj', (object,), {'host': '0.0.0.0', 'port': 8000})()
        log_config = None
    return ConfigClass()

def create_app(config):
    app = FastAPI()

    # 1. 先创建SchemaNames
    schema_names = create_schema_names(config.prefix)
    # 2. 再用SchemaNames生成模型
    raw_model, staging_model = create_models(schema_names)

    # 3. 注册API，把模型传递给需要的路由
    register_api_1(app, config)
    register_api_2(app, raw_model, staging_model)

    return app

def main():
    config = make_config()
    app = create_app(config)
    uvicorn.run(
        app,
        host=config.webserver.host,
        port=config.webserver.port,
        log_config=config.log_config,
    )

if __name__ == "__main__":
    main()

这个方案的优势：完全没有循环依赖，没有全局变量，所有依赖都是显式传递的，符合依赖注入的最佳实践，而且测试时可以很容易传入不同的SchemaNames来验证不同环境的行为。

方案二：延迟导入+动态表参数（适合不想重构太多代码的场景） #

如果不想大改模型结构，可以用延迟导入模型+declared_attr动态生成表参数的方式，绕开导入时的依赖问题。

步骤1：调整enums.py为可动态设置的占位符 #

from enum import Enum, Type

# 先初始化一个占位符，后续在main.py中设置
SchemaNames = None

def create_schema_names(prefix: str) -> Type[Enum]:
    global SchemaNames
    class TempSchemaNames(str, Enum):
        RAW = prefix + "_RAW"
        STAGING = prefix + "_STAGING"
        TRANSFORMED = prefix + "_TRANSFORMED"
    SchemaNames = TempSchemaNames
    return SchemaNames

步骤2：修改模型用declared_attr动态获取schema #

使用SQLAlchemy的declared_attr，让__table_args__在模型实际被使用时才求值，而不是导入时：

from sqlalchemy import Column, String, Integer
from sqlalchemy.ext.declarative import declarative_base, declared_attr
from app.utils.enums import SchemaNames

Base = declarative_base()

class RawData(Base):
    __tablename__ = "raw_data"
    
    @declared_attr
    def __table_args__(cls):
        return {"schema": SchemaNames.RAW.value}
    
    id = Column(Integer, primary_key=True)
    data = Column(String)

class StagingData(Base):
    __tablename__ = "staging_data"
    
    @declared_attr
    def __table_args__(cls):
        return {"schema": SchemaNames.STAGING.value}
    
    id = Column(Integer, primary_key=True)
    processed_data = Column(String)

步骤3：在main.py中先初始化Enum再导入API #

from fastapi import FastAPI
from app.api import register_api_1
from app.utils.enums import create_schema_names
import uvicorn

def make_config():
    # 模拟配置解析
    class ConfigClass:
        prefix = "MY_APP"
        webserver = type('obj', (object,), {'host': '0.0.0.0', 'port': 8000})()
        log_config = None
    return ConfigClass()

def create_app(config):
    app = FastAPI()

    # 1. 先初始化SchemaNames
    create_schema_names(config.prefix)

    # 2. 现在SchemaNames已经就绪，再导入并注册API
    register_api_1(app, config)
    from app.api import register_api_2
    register_api_2(app)

    return app

def main():
    config = make_config()
    app = create_app(config)
    uvicorn.run(
        app,
        host=config.webserver.host,
        port=config.webserver.port,
        log_config=config.log_config,
    )

if __name__ == "__main__":
    main()

这个方案用了一个受控的全局变量（只在初始化时设置一次，之后只读），虽然不是完全无全局，但代码改动小，适合快速调整。

最佳实践总结 #

• 优先选方案一：显式依赖传递，代码更清晰、可维护性和可测试性更强，完全符合现代Python项目的依赖管理规范。
• 如果项目已经有大量现有模型，不想重构，可以用方案二过渡，但长期来看还是推荐逐步迁移到工厂模式。

备注：内容来源于stack exchange，提问作者user3557405