我在 Alembic + FastAPI + PostgreSQL 上遇到这个问题
SQL Alchemy Alembic 反向生成迁移，降级创建表并升级删除表，而数据库为空。
例如：
alembic revision --autogenerate -m "Testing 2"
INFO [alembic.runtime.migration] Context impl PostgresqlImpl.
INFO [alembic.runtime.migration] Will assume transactional DDL.
INFO [alembic.autogenerate.compare] Detected removed index 'ix_Site_keyword' on 'Site'
INFO [alembic.autogenerate.compare] Detected removed table 'Site'
Generating /app/migrations/versions/1e3a0f40182c_testing_2.py ... done
我检查了文件，得到了这个：
"""Testing 2
Revision ID: 1e3a0f40182c
Revises: 928ab2a61fa7
Create Date: 2024-04-04 08:32:29.784316
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects import postgresql
# revision identifiers, used by Alembic.
revision: str = '1e3a0f40182c'
down_revision: Union[str, None] = '928ab2a61fa7'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
op.drop_index('ix_Site_keyword', table_name='Site')
op.drop_table('Site')
# ### end Alembic commands ###
def downgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
op.create_table('Site',
sa.Column('id', sa.UUID(), autoincrement=False, nullable=False),
sa.Column('title', sa.VARCHAR(), autoincrement=False, nullable=True),
sa.Column('keyword', sa.VARCHAR(), autoincrement=False, nullable=True),
sa.Column('description', sa.TEXT(), autoincrement=False, nullable=True),
sa.Column('is_active', sa.BOOLEAN(), autoincrement=False, nullable=True),
sa.Column('created_at', postgresql.TIMESTAMP(), autoincrement=False, nullable=True),
sa.Column('updated_at', postgresql.TIMESTAMP(), autoincrement=False, nullable=True),
sa.Column('deleted_at', postgresql.TIMESTAMP(), autoincrement=False, nullable=True),
sa.PrimaryKeyConstraint('id', name='Site_pkey')
)
op.create_index('ix_Site_keyword', 'Site', ['keyword'], unique=False)
# ### end Alembic commands ###
为什么要做这个？如果我切换块（将升级块放入降级块，反之亦然）如果我运行下一个自动迁移，我会在升级功能中得到所有删除语法......所以，我总是必须重写代码。
这是怎么回事？
更新：这是我的 env.py 文件。
from logging.config import fileConfig
from sqlalchemy import engine_from_config
from sqlalchemy import pool
from database.orm import Base
from alembic import context
config = context.config
if config.config_file_name is not None:
fileConfig(config.config_file_name)
target_metadata = Base.metadata
def run_migrations_offline() -> None:
url = config.get_main_option("sqlalchemy.url")
context.configure(
url=url,
target_metadata=target_metadata,
literal_binds=True,
dialect_opts={"paramstyle": "named"},
)
with context.begin_transaction():
context.run_migrations()
def run_migrations_online() -> None:
connectable = engine_from_config(
config.get_section(config.config_ini_section, {}),
prefix="sqlalchemy.",
poolclass=pool.NullPool,
)
with connectable.connect() as connection:
context.configure(
connection=connection, target_metadata=target_metadata
)
with context.begin_transaction():
context.run_migrations()
if context.is_offline_mode():
run_migrations_offline()
else:
run_migrations_online()
最后：
env.py
from database.orm import Base
from models import site
models.site.py
from sqlalchemy import Boolean, Column, String, DateTime, Text
from database.orm import Base
import uuid
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy import func
def generate_uuid():
return str(uuid.uuid4())
class Site(Base):
__tablename__ = "Site"
id = Column(UUID(as_uuid=True), primary_key=True, unique=True, default=generate_uuid)
title = Column(String)
keyword = Column(String, index=True)
description = Column(Text)
is_active = Column(Boolean, default=True)
created_at = Column(DateTime, default=func.now)
updated_at = Column(DateTime, default=func.now)
deleted_at = Column(DateTime, default=func.now)
数据库.orm.py
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
SQLALCHEMY_DATABASE_URL = "postgresql://postgres:example@database/db_new"
engine = create_engine(
SQLALCHEMY_DATABASE_URL
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
运行：
alembic revision --autogenerate -m "Tes4"
INFO [alembic.runtime.migration] Context impl PostgresqlImpl.
INFO [alembic.runtime.migration] Will assume transactional DDL.
INFO [alembic.autogenerate.compare] Detected added table 'Site'
INFO [alembic.autogenerate.compare] Detected added index ''ix_Site_keyword'' on '('keyword',)'
Generating /app/migrations/versions/f3d6623e75ec_tes4.py ... done
迁移.versions.f3d6623e75ec_tes4.py
"""Tes4
Revision ID: f3d6623e75ec
Revises:
Create Date: 2024-04-04 15:19:09.018548
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = 'f3d6623e75ec'
down_revision: Union[str, None] = None
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
op.create_table('Site',
sa.Column('id', sa.UUID(), nullable=False),
sa.Column('title', sa.String(), nullable=True),
sa.Column('keyword', sa.String(), nullable=True),
sa.Column('description', sa.Text(), nullable=True),
sa.Column('is_active', sa.Boolean(), nullable=True),
sa.Column('created_at', sa.DateTime(), nullable=True),
sa.Column('updated_at', sa.DateTime(), nullable=True),
sa.Column('deleted_at', sa.DateTime(), nullable=True),
sa.PrimaryKeyConstraint('id'),
sa.UniqueConstraint('id')
)
op.create_index(op.f('ix_Site_keyword'), 'Site', ['keyword'], unique=False)
# ### end Alembic commands ###
def downgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
op.drop_index(op.f('ix_Site_keyword'), table_name='Site')
op.drop_table('Site')
# ### end Alembic commands ###