FastAPI+SQLAlchemy环境下Alembic迁移未生成建表代码求助

问题概述 #

使用FastAPI和SQLAlchemy构建含多对多关系的数据库时，执行Alembic迁移生成命令后，得到的迁移文件仅包含空的upgrade()和downgrade()函数，重装Alembic、创建新项目均无法解决。

核心原因 #

1. 元数据对象不匹配：env.py中指定的target_metadata = metadata是单独创建的空MetaData实例，但所有模型类都继承自declarative_base()生成的Base，模型的元数据实际存储在Base.metadata中，导致Alembic无法识别模型结构。
2. 模型关系定义错误：Material与MaterialType的关系配置错误，两者应为一对多关系，却误用了多对多的secondary参数；Coffin与Material的多对多关系未配置双向关联。
3. 中间表主键冗余：CoffinMaterial中同时用primary_key=True和PrimaryKeyConstraint定义主键，造成冗余。
4. 表创建函数错误：create_tables中调用metadata.create_all，实际应使用Base.metadata.create_all来创建模型对应的表。

解决方案 #

1. 修正env.py的元数据引用 #

将env.py中的target_metadata改为Base.metadata，确保Alembic能读取到所有模型定义：

# 替换原导入和target_metadata配置
from app.api.db_model import Base, DATABASE_URL  # 导入Base而非metadata

# ...

target_metadata = Base.metadata  # 使用Base的元数据

2. 修复模型关系与定义 #

修正模型中的关系配置和主键定义：

from sqlalchemy import (Column, Integer, String, ForeignKey, PrimaryKeyConstraint)
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker, declarative_base, relationship
import asyncio

DATABASE_URL = "postgresql+asyncpg://myuser:mypassword@localhost:myport/mydatabase"
engine = create_async_engine(DATABASE_URL)
AsyncSessionLocal = sessionmaker(
    bind=engine, class_=AsyncSession, expire_on_commit=False
)

Base = declarative_base()


class Material(Base):
    __tablename__ = "materials"
    id_material = Column(Integer, primary_key=True, index=True)
    name = Column(String)
    id_type = Column(Integer, ForeignKey("material_types.id_type"))
    # 一对多关系：Material属于一个MaterialType
    material_type = relationship("MaterialType", back_populates="materials")
    # 多对多关系：Material关联多个Coffin
    coffins = relationship("Coffin", secondary="coffin_materials", back_populates="materials")


class MaterialType(Base):
    __tablename__ = "material_types"
    id_type = Column(Integer, primary_key=True, index=True)
    name = Column(String)
    # 一对多反向关联：MaterialType包含多个Material
    materials = relationship("Material", back_populates="material_type")


class Coffin(Base):
    __tablename__ = "coffins"
    id_coffin = Column(Integer, primary_key=True, index=True)
    name = Column(String)
    price = Column(Integer)
    length = Column(Integer)
    width = Column(Integer)
    quantity = Column(Integer)
    # 多对多反向关联：Coffin包含多个Material
    materials = relationship("Material", secondary="coffin_materials", back_populates="coffins")


class CoffinMaterial(Base):
    __tablename__ = "coffin_materials"
    id_coffin = Column(Integer, ForeignKey("coffins.id_coffin"), primary_key=True)
    id_material = Column(Integer, ForeignKey("materials.id_material"), primary_key=True)
    # 移除冗余的PrimaryKeyConstraint，因为已通过primary_key=True定义复合主键


async def create_tables():
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)  # 使用Base.metadata

# asyncio.run(create_tables())  # 建议仅在初始化时执行，避免重复运行

from databases import Database
database = Database(DATABASE_URL)

3. 重新生成迁移文件 #

执行以下命令重新生成迁移：

alembic revision --autogenerate -m "first_migration"
alembic upgrade head

验证结果 #

修正后，生成的迁移文件将包含所有表的创建语句，包括多对多中间表的结构，执行upgrade后数据库会正确创建所有表。

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