如何通过代码使用 Python Alembic 运行迁移?

本文介绍了如何通过代码使用 Python Alembic 运行迁移?的处理方法，对大家解决问题具有一定的参考价值，需要的朋友们下面随着小编来一起学习吧！

问题描述

我正在尝试使用 sqlAlchemy 和 Alembic 迁移 SQL 数据库.我想直接在 Python 代码中制作简单的迁移脚本，而不是使用 Alembic CLI，如文档中所述.

I am trying to migrate a SQL database using sqlAlchemy and Alembic.I would like to make simple migration scripts directly in Python code, and not by using Alembic CLI, as described in the docs.

我只在这个主题上找到了这个 SO 问题:使用来自应用程序代码内部的 Alembic API

I only found this SO question on this topic: Using Alembic API from inside application code

使用这个问题 + Flask-Alembic 源代码，我尝试了这些简单的命令.引擎链接到我的 db &元数据包含迁移信息.

Using this question + Flask-Alembic source code, I tried these simple commands.engine is linked to my db & metadata contains the information for the migration.

我认为我非常接近，解决方案应该在一行代码中......我很挣扎.

I think I am very close, and the solution should be in one line of code... I'm struggling.

from alembic.config import Config
from alembic import command, autogenerate
from alembic.script import ScriptDirectory
from alembic.runtime.environment import EnvironmentContext

alembic_cfg = Config()
alembic_cfg.set_main_option("script_location", "migrations")
alembic_cfg.set_main_option("url", "postgresql://user:pass@postgres:5432/mydb")

alembic_script = ScriptDirectory.from_config(alembic_cfg)
alembic_env = EnvironmentContext(alembic_cfg, alembic_script)

conn = engine.connect()
alembic_env.configure(connection=conn, target_metadata=metadata)
alembic_context = alembic_env.get_context()

我可以使用以下命令来查看它是否有效并检测必须迁移哪些字段:

I am able to use the following commands to see that it works and detects what fields have to migrate :

autogenerate.compare_metadata(alembic_context, metadata)
autogenerate.produce_migrations(alembic_context, metadata)

但是，我无法运行迁移.我试了几个命令，总是报错...

However, I am not able to run the migrations. I tried several commands, and always get an error...

例如，如果我跑:

with alembic_env.begin_transaction():
    alembic_env.run_migrations()

我明白了:

/usr/local/lib/python2.7/site-packages/alembic/runtime/migration.pyc in run_migrations(self, **kw)
    301         head_maintainer = HeadMaintainer(self, heads)
    302
--> 303         for step in self._migrations_fn(heads, self):
    304             with self.begin_transaction(_per_migration=True):
    305                 if self.as_sql and not head_maintainer.heads:

TypeError: 'NoneType' object is not callable

谁能帮帮我.我认为解决方案是单行的，但我找不到如何在我的数据库上运行迁移...

Could anyone help me. I think the solution is a one-liner, but I cannot find how to run the migration on my database...

有关信息，我从未使用 Alembic 对这个数据库做过任何迁移，也许需要一个init"?

For information, I have never done any migration on this DB with Alembic, maybe there is the need of an "init" ?

非常感谢.

推荐答案

for step in self._migrations_fn(heads, self):

这里我们将描述您要迁移的顺序.该序列由步骤组成，每个步骤都有 step.migration_fn(**kw).

Here we shall describe in what order you want to migrate. The sequence consists of steps, each of have step.migration_fn(**kw).

您需要的最后一步是在执行 alembic_env.configure 时添加 _migrations_fn.

The last step you need is add _migrations_fn while you do alembic_env.configure.

def do_upgrade(revision, context):
    return alembic_script._upgrade_revs(script.get_heads(), revision)

alembic_env.configure(connection=conn, target_metadata=metadata, fn=do_upgrade)

script.get_heads() 返回最后一次迁移.替换，如果你需要一个特定的修订，而不是最后一个.

script.get_heads() returns the last migration. Replace, if you need a specific revision, instead of the last.