以编程方式获取 Alembic 数据库版本答案

【问题标题】：Getting Alembic Database Version Programmatically以编程方式获取 Alembic 数据库版本
【发布时间】：2014-08-17 08:32:54
【问题描述】：

我正在尝试弄清楚如何使用 Alembic 获取我的数据库版本。我已经将数据库设置为使用 alembic 并成功地对其进行了升级和降级。我现在想从我自己的 python 脚本中获取这个版本。

我试图为此创建一个函数

def get_current_database_version():
    path = os.path.join(os.path.dirname(__file__), os.path.pardir)
    alembic_cfg = Config(os.path.join(path, 'alembic.ini'))
    current_rev = command.current(alembic_cfg, head_only=True)
    return current_rev

这个函数返回一个NoSectionError: No section: 'formatters'

然后我转到我的 alembic.ini 文件以检查它是否有格式化程序区域。这是我的 alembic.ini 文件：

# A generic, single database configuration.

[alembic]
# path to migration scripts
script_location = alembic
pyramid_config_file = ../../development.ini

# template used to generate migration files
# file_template = %%(rev)s_%%(slug)s

# max length of characters to apply to the
# "slug" field
#truncate_slug_length = 40

# set to 'true' to run the environment during
# the 'revision' command, regardless of autogenerate
# revision_environment = false

# set to 'true' to allow .pyc and .pyo files without
# a source .py file to be detected as revisions in the
# versions/ directory
# sourceless = false

sqlalchemy.url = sqlite:///%(here)s/mgo.sqlite


# Logging configuration
[loggers]
keys = root,sqlalchemy,alembic

[handlers]
keys = console

[formatters]
keys = generic

[logger_root]
level = WARN
handlers = console
qualname =

[logger_sqlalchemy]
level = WARN
handlers =
qualname = sqlalchemy.engine

[logger_alembic]
level = INFO
handlers =
qualname = alembic

[handler_console]
class = StreamHandler
args = (sys.stderr,)
level = NOTSET
formatter = generic

[formatter_generic]
format = %(levelname)-5.5s [%(name)s] %(message)s
datefmt = %H:%M:%S

有人知道我做错了什么吗？谢谢

编辑：

这是我尝试使用 MigrationContext 解决问题：

def get_database_revision():
    engine = create_engine("sqlite:///../mgo.db")
    conn = engine.connect()
    context = MigrationContext.configure(conn)
    current_rev = context.get_current_revision()
    return current_rev

它连接但没有返回。使用sqlite浏览器可以看到数据库中的版本没有设置为none。

【问题讨论】：

标签： python sqlalchemy alembic

【解决方案1】：

您可以使用MigrationContext 到get the current version：

from alembic.migration import MigrationContext
from sqlalchemy import create_engine

engine = create_engine("postgresql://mydatabase")
conn = engine.connect()

context = MigrationContext.configure(conn)
current_rev = context.get_current_revision()

你可以在env.py里面使用：

from alembic import context
migration_context = context.get_context()
current_rev = context.get_current_revision()

最后，它基本上归结为连接到数据库并查看alembic_version 表。它包含迁移版本作为值，这就是数据库当前所在的位置（根据 alembic）。所以你可以用任何你想要的方式编写代码，只要那是你最终要做的事情。

【讨论】：

我改变了我的代码来尝试这个，它也没有工作。它返回无。
你检查你是否正确连接到数据库？
请注意，在从来没有版本的 alembic 中，MigrationContext 被移动到 alembic.runtime.migration

【解决方案2】：

我建议使用 stdout Config() 对象参数（请参阅here）来允许将 sys.stdout 重定向到 StringIO 缓冲区，如此处所做的：

output_buffer = io.StringIO()
alembic_cfg = alembic_Config('/path/to/alembic.ini', stdout=output_buffer)
alembic_command.current(alembic_cfg)
output = output_buffer.getvalue()
print(output)

【讨论】：

【解决方案3】：

这个问题很老，但我有一个我认为比目前给出的更简单的解决方案。

主要观察是当command.current被调用时，alembic没有使用Python内置的print函数，而是在config对象上使用print_stdout方法。因此，要捕获输出，只需重载print_stdout 函数！这对我有用：

def get_current_database_version():
    path = os.path.join(os.path.dirname(__file__), os.path.pardir)
    alembic_cfg = Config(os.path.join(path, 'alembic.ini'))

    captured_text = None
    def print_stdout(text, *arg):
        nonlocal captured_text
        captured_text = text

    alembic_cfg.print_stdout = print_stdout
    command.current(alembic_cfg)
    return captured_text

【讨论】：

我只想指出这适用于其他命令，例如：command.heads 和 command.history

【解决方案4】：

只是想投入我的 2 美分。

首先，MigrationContext 可能对您不起作用，因为您没有连接到正确的数据库。从我看到的文档中，如果找不到您指定的文件，create_engine 将为您创建一个数据库。这可能是因为在您的示例中，您使用的相对路径很容易混淆。

其次，最让我痛心的是

command.current(alembic_cfg, head_only=True)

实际上只有显示当前版本，对我来说，这似乎是在 Eclipse 中将值打印到控制台。函数本身总是返回 None 这有点烦人，这就是需要 MigrationContext 的原因。

此外，如果您检查当前版本是因为您想知道数据库的状态而不实际更新它，那么您将需要使用 ScriptDirectory http://alembic.readthedocs.org/en/latest/api/script.html#alembic.script.ScriptDirectory 以及它的各种方法来确定 MigrationContext 返回的版本是当前头还是完全有效。

【讨论】：

【解决方案5】：

只是建立在James Fennel's 答案上，在获取历史记录的情况下，它只会输出一行。如果您想要所有输出（作为列表），您可以这样做：

def get_current_database_history(input_cfg, start_rev=None):
    captured_text = []
    def print_stdout(text, *arg):
        nonlocal captured_text
        captured_text.append(text)
    input_cfg.print_stdout = print_stdout
    if start_rev:
        command.history(input_cfg, rev_range="{}:".format(start_rev))
    else:
        command.history(input_cfg)
    return captured_text

【讨论】：

【解决方案6】：

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


class DBMigrations:

    def __init__(self):
        self.alembic_cfg = Config("./alembic.ini")
        self.alembic_cfg.set_main_option('sqlalchemy.url', DATABASE_URL)
        self.script = ScriptDirectory.from_config(self.alembic_cfg)

    def get_db_version(self):
        current_version = ""

        def display_version(rev, context):
            for rev in self.script.get_all_current(rev):
                nonlocal current_version
                current_version = rev.cmd_format(verbose=False)
            return []

        with EnvironmentContext(self.alembic_cfg, self.script, fn=display_version, dont_mutate=True):
            self.script.run_env()
        return current_version.split()[0]

【讨论】：

【解决方案7】：

这是一个返回正确版本号的临时修复。

def get_database_revision():
    s = select(['version_num'],from_obj='alembic_version')
    result = DBSession.execute(s).fetchone()
    return result['version_num']

如果没有找到版本号，此函数返回 None。

【讨论】：

现在我们不得不使用这个版本的表达式select([text('version_num')], from_obj=text('alembic_version'))
导入哪个模块来获得select？ :)

【解决方案8】：

一个黑客！在正确的目录中使用sh，就这么简单：

>>> from sh import alembic
>>> alembic("current")
4cad21a83709 (head)

【讨论】：

嗯，可以很容易地以同样的方式使用 subprocess，但我不认为这就是问题所在。
当然取决于我们（或 OP）如何定义“程序化”！该术语是否应该明确排除调用另一个进程？

【解决方案9】：

我刚刚阅读了 alembic_version 表的数据库：

    try:
        version_num = db.execute("SELECT version_num FROM alembic_version").first()[0]
    except:
        version_num = None

【讨论】：