SQLAlchemy + alembic：创建模式迁移答案

【问题标题】：SQLAlchemy + alembic: create schema migrationSQLAlchemy + alembic：创建模式迁移
【发布时间】：2018-09-09 22:38:30
【问题描述】：

我不确定如何定义create schema foo 迁移？我的模型看起来像这样（我正在使用 Flask-Migrate）：

class MyTable(db.Model):
    __tablename__ = "my_table"
    __table_args__ = {"schema": "foo"}

    id = ColumnPrimKey()
    name = Column(Text, unique=True, nullable=False)

当我执行mange db upgrade 时，由于模式“foo”不存在而失败。如何使用 SQLAlchemy 和 Alembic 为架构添加迁移？

【问题讨论】：

标签： sqlalchemy alembic flask-migrate

【解决方案1】：

我通过将迁移upgrade 命令修改为首次运行来完成此操作：

op.execute("create schema foo")

在downgrade函数中

op.execute("drop schema foo")

所以整个迁移文件看起来像这样：

from alembic import op
import sqlalchemy as sa


# revision identifiers, used by Alembic.
revision = '6c82c972c61e'
down_revision = '553260b3e828'
branch_labels = None
depends_on = None


def upgrade():
    op.execute("create schema foo")
    ...

def downgrade():
    ...
    op.execute("drop schema foo")

【讨论】：

有什么方法可以让sqlalchemy.event 工作吗？例如，event.listen(Base.metadata, "before_create", CreateSchema("foo"))。
这解决了迁移问题，但是如果您有另一个迁移，它不会再次检测到您的架构并会生成所有表......这可能很麻烦。有没有办法检测不同架构中的变化？
@Pstr 您可以在元数据类上指定架构属性metadata = Metadata(schema=<schema-name>) 或Base = declarative_base(schema=<schema-name>)
作为对我自己的问题的回答，如果您在 alembic/env.py 文件的 context.configure() 函数中设置参数 include_schemas=True,，alembic 将确认模式。不知道为什么这个设置没有默认开启

【解决方案2】：

另一种选择是添加以下函数来修改env.py 中的 MigrationScript 指令：

from alembic import operations

def process_revision_directives(context, revision, directives):
    """Modify the MigrationScript directives to create schemata as required.
    """
    script = directives[0]
    for schema in frozenset(i.schema for i in target_metadata.tables.values()):
        script.upgrade_ops.ops.insert(
            0, operations.ops.ExecuteSQLOp(f"CREATE SCHEMA IF NOT EXISTS {schema}"))
        script.downgrade_ops.ops.append(
            operations.ops.ExecuteSQLOp(f"DROP SCHEMA IF EXISTS {schema} RESTRICT"))

然后在context.configure 中添加process_revision_directives=process_revision_directives。

【讨论】：

【解决方案3】：

已接受答案的一个潜在问题是，对于初始迁移，Alembic 可能无法找到创建alembic_version 的位置。这是因为op.execute("create schema foo") 仅在Alembic 尝试查找其alembic_version 表之后执行。错误弹出为：

sqlalchemy.exc.ProgrammingError: (psycopg2.errors.InvalidSchemaName) schema "foo" does not exist

简单的方法是让alembic_version 表存在于另一个架构中，将version_table_schema 传递给context.configure() (docs)。

但是，在许多情况下，您可能希望同时 (i) 从头开始创建一个模式作为初始迁移的一部分（例如设置测试环境）和 (ii) 在目标模式中拥有 alembic_version 表.在这些情况下，另一种方法是将架构创建委托给env.py。示例：

# env.py
from sqlalchemy import create_engine

from bar import create_database_with_schema_if_not_exists

SQLALCHEMY_DATABASE_URI = ...
schema = "foo"
engine = create_engine(SQLALCHEMY_DATABASE_URI)
create_database_with_schema_if_not_exists(engine, schema)
...

# bar.py
import sqlalchemy
from sqlalchemy_utils import create_database, database_exists

def create_database_with_schema_if_not_exists(engine, schema):
    if not database_exists(engine.url):
        create_database(engine.url)
    if not engine.dialect.has_schema(engine, schema):
        engine.execute(sqlalchemy.schema.CreateSchema(schema))

【讨论】：