Alembic无法在PostgreSQL Schema生成表及事务未提交问题排查

核心问题分析 #

你的问题根源是事务未正确提交，加上Alembic迁移时的连接/事务处理逻辑冲突，导致所有数据库修改无法持久化：

• Python会话中执行SQL后，事务未提交，只有当前会话能看到临时变更，其他会话（如psql CLI）看不到，会话终止后变更自动回滚
• Alembic迁移时的事务嵌套逻辑，导致迁移操作执行后被隐式回滚，无法生成版本表或修改数据库

分步解决方案 #

1. 修复Python执行SQL的持久化问题 #

你的DatabaseSession中sessionmaker默认autocommit=False，所有操作都在未提交的事务中运行，必须手动提交才能让变更生效：

修改_connect_db函数（显式指定schema+优化事务逻辑）

def _connect_db(self, db_schema_override: str = None):
    schema = self.schema if not db_schema_override else db_schema_override
    try:
        engine = create_engine(
            self.db_url, 
            connect_args={"options": "-csearch_path={}".format(schema)}
        )
        SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
        session = SessionLocal()
        # 显式绑定metadata到目标schema，避免反射错误
        metadata = MetaData(schema=schema)
        metadata.reflect(bind=engine)
        metadata.create_all(engine)
        return engine, session
    except (AttributeError, ValueError):
        raise

执行SQL时手动提交事务

# 示例：执行CREATE TABLE后必须提交
engine, session = db._connect_db()
session.execute(text("CREATE TABLE test (id INT)"))
session.commit()  # 关键：手动提交事务
session.close()

2. 修复Alembic迁移无效果的问题 #

修改upgrade_db函数（取消事务嵌套）

不要自己创建连接事务，让Alembic自行管理连接和事务：

def upgrade_db(db_schema: str, revision: str = "head") -> None:
    db = DatabaseSession(db_schema)
    db.db_data.maintenance_mode = True
    db.db_data.save()
    _config = config.Config("path/to/file/alembic.ini")
    _config.set_main_option("sqlalchemy.url", db.db_url)
    _config.attributes["schema"] = db_schema.lower()
    # 直接调用upgrade，让Alembic自己处理连接
    command.upgrade(_config, revision)
    db.db_data.maintenance_mode = False
    db.db_data.save()

修改run_migrations_online函数（确保事务正确提交+schema配置）

def run_migrations_online():
    connectable = engine_from_config(
        config.get_section(config.config_ini_section),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )
    db_schema = config.attributes.get("schema", "public")

    # 用connectable.begin()管理顶级事务，自动提交
    with connectable.begin() as connection:
        connection.execute(text(f'CREATE SCHEMA IF NOT EXISTS "{db_schema}"'))
        connection.execute(text(f"SET search_path TO '{db_schema}'"))
        connection.dialect.default_schema_name = db_schema

        context.configure(
            connection=connection,
            target_metadata=target_metadata,
            include_schemas=True,
            version_table_schema=db_schema,  # 强制Alembic版本表生成在目标schema中
        )

        # 无需再开嵌套事务，直接执行迁移
        context.run_migrations()

3. 验证连接字符串（可选） #

确保PostgreSQL连接字符串正确指定schema，避免连接时的默认路径错误：

db_url = "postgresql://user:password@postgres:5432/dbname?options=-csearch_path=your_target_schema"

验证步骤 #

1. 先在Python中执行CREATE TABLE并提交，用psql CLI执行\dt your_schema.*确认表存在
2. 执行alembic upgrade head后，检查目标schema中是否生成alembic_version表
3. 若仍有问题，开启SQLAlchemy日志查看执行细节：

engine = create_engine(
    db_url,
    connect_args={"options": "-csearch_path={}".format(schema)},
    echo=True  # 打印所有SQL语句，排查是否有提交操作
)

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