0


我正在使用 Async FastApi、SQLAlchemy、Alembic 和 Docker。我有两个数据库。其中一个是完全空的，没有一张桌子。另一个有我需要在我的 FastApi 应用程序中使用的所有表。我不想使用像这样的 alembic 和手动运行脚本进行手动迁移：

管理.py

import asyncio

from database import Base, engine


async def init_models():
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all, checkfirst=True)


if __name__ == "__main__":
    asyncio.run(init_models())

我想在docker启动时运行所有这些，所以我这样做了：

入口点.sh

echo "Running migrations"
alembic upgrade head
python manage.py

这是我的database.py文件。我有数据库设置、一些表和我用sqlalchemy_utils创建的一个视图。

数据库.py

from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from sqlalchemy.orm.decl_api import DeclarativeMeta
from sqlalchemy_utils import create_view

from confing imnport settings


engine = create_async_engine(settings.DATABSE_URL, pool_pre_ping=True)
Base: DeclarativeMeta = declarative_base()
async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)


profiles = (
    select(
        [
            ProfileInstance.id,
            ProfileInstance.title,
            ProfileInstance.created_at,
            ProfileVersion.changed_at,
        ]
    )
    .select_from(
        ProfileInstance.__table__.join(
            ProfileVersion,
            ProfileVersion.instance_id == ProfileInstance.id,
        )
    )
)

profile_view = create_view("fastapi.profiles", profiles, Base.metadata)


class Profiles(Base):
    __table_args__ = {"schema": "fastapi"}
    __tablename__ = "profiles"
    __table__ = profile_view 


async def get_session() -> AsyncSession:
    async with async_session() as session:
        yield session

这是我的问题。
当我在 empy DB 上运行manage.py时，它运行良好。但是当我在完整的数据库上运行它时，它会引发一个错误：

<class 'asyncpg.exceptions.DuplicateTableError'>：关系“profiles”已经存在

我尝试使用该问题中的 RenaKunisaki 评论。它适用于完整的数据库，没有错误。但是在空数据库上它不会创建我的视图（

我尝试像sqlalchemy docs上所说的那样创建我的视图。但我在

assert profile_view.primary_key == [profile_view.id]

# Or

assert profile_view.primary_key == [profile_view.c.id]

我没有任何身份证

我该如何处理？我会很感激任何答案

4

0 回答 0