如何在不生成迁移脚本的情况下迁移 SQLAlchemy 数据库
How to migrate SQLAlchemy database without generating migration script
我在无服务器函数中使用 Flask 和 SQLAlchemy,但我无权访问命令行。有时我想更改 tables。基本上我想自动创建新的tables(已经由create_all
完成)和列,以及更改列(例如更长的字符串)。
如何在命令行不生成迁移脚本的情况下迁移SQLAlchemy数据库?如果我能回顾一下要完成的任务(例如[添加A列,删除table B])会更好。
是的,使用 alembic 是可能的,而且不是很 hacky。
我使用蒸馏器 autogenerate API。 compare_metadata
方便生成diff进行review,produce_migrations
方便生成迁移操作,由Opertaions.invoke
调用。
需要注意的一件事是 upgrade_ops
可能包含 OpContainer
,它本身不能 invoke
,但具有包含嵌套操作的 ops
属性。
下面是我实现这个的代码:
@bp.route("/migrate", methods=["GET", "POST"])
def migrate():
from alembic.runtime.migration import MigrationContext
# use `db.session.connection()` instead of `db.engine.connect()`
# to avoid database lock hang
context = MigrationContext.configure(db.session.connection())
if request.method == "GET":
import pprint
from alembic.autogenerate import compare_metadata
diff = compare_metadata(context, db.metadata)
diff_str = pprint.pformat(diff, indent=2, width=20)
logger.info("Migrate steps: %s", diff_str)
return respond_success(migration=diff_str)
from alembic.autogenerate import produce_migrations
from alembic.operations import Operations
from alembic.operations.ops import OpContainer
migration = produce_migrations(context, db.metadata)
operation = Operations(context)
for outer_op in migration.upgrade_ops.ops:
logger.info("Invoking %s", outer_op)
if isinstance(outer_op, OpContainer):
for inner_op in outer_op.ops:
logger.info("Invoking %s", inner_op)
operation.invoke(inner_op)
else:
operation.invoke(outer_op)
return respond_success()
对于阅读本文的任何人,如果您想确保审查的操作与要调用的操作相同,请尝试发布 diff_str
的哈希值并进行比较。
我在无服务器函数中使用 Flask 和 SQLAlchemy,但我无权访问命令行。有时我想更改 tables。基本上我想自动创建新的tables(已经由create_all
完成)和列,以及更改列(例如更长的字符串)。
如何在命令行不生成迁移脚本的情况下迁移SQLAlchemy数据库?如果我能回顾一下要完成的任务(例如[添加A列,删除table B])会更好。
是的,使用 alembic 是可能的,而且不是很 hacky。
我使用蒸馏器 autogenerate API。 compare_metadata
方便生成diff进行review,produce_migrations
方便生成迁移操作,由Opertaions.invoke
调用。
需要注意的一件事是 upgrade_ops
可能包含 OpContainer
,它本身不能 invoke
,但具有包含嵌套操作的 ops
属性。
下面是我实现这个的代码:
@bp.route("/migrate", methods=["GET", "POST"])
def migrate():
from alembic.runtime.migration import MigrationContext
# use `db.session.connection()` instead of `db.engine.connect()`
# to avoid database lock hang
context = MigrationContext.configure(db.session.connection())
if request.method == "GET":
import pprint
from alembic.autogenerate import compare_metadata
diff = compare_metadata(context, db.metadata)
diff_str = pprint.pformat(diff, indent=2, width=20)
logger.info("Migrate steps: %s", diff_str)
return respond_success(migration=diff_str)
from alembic.autogenerate import produce_migrations
from alembic.operations import Operations
from alembic.operations.ops import OpContainer
migration = produce_migrations(context, db.metadata)
operation = Operations(context)
for outer_op in migration.upgrade_ops.ops:
logger.info("Invoking %s", outer_op)
if isinstance(outer_op, OpContainer):
for inner_op in outer_op.ops:
logger.info("Invoking %s", inner_op)
operation.invoke(inner_op)
else:
operation.invoke(outer_op)
return respond_success()
对于阅读本文的任何人,如果您想确保审查的操作与要调用的操作相同,请尝试发布 diff_str
的哈希值并进行比较。