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

Question

我正在尝试用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文件，看看里面是否有formatters这个部分。以下是我的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

它连接上了，但返回的是none。通过sqlite浏览器我可以看到数据库中的版本并没有设置为none。

Answer 1

我想分享一下我的看法。

首先，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返回的版本是否是当前的最新版本，或者它是否有效。

Answer 2

在编程中，有时候我们会遇到一些问题，特别是在使用某些工具或库的时候。比如，有人可能会在使用某个特定的功能时，发现它的表现和预期不一样。这种情况可能会让人感到困惑，因为我们总是希望我们的代码能够顺利运行。

解决这类问题的第一步是仔细检查代码，看看是否有拼写错误或者逻辑上的问题。很多时候，问题可能就藏在这些小细节里。

另外，查看相关的文档也是一个好主意。文档通常会提供使用某个功能的详细说明，帮助我们理解如何正确使用它。

如果自己无法解决问题，可以考虑向社区求助，比如在StackOverflow上提问。在提问时，记得把你的代码和遇到的问题描述清楚，这样别人才能更好地帮助你。

总之，遇到问题时不要着急，慢慢分析，查阅资料，必要时寻求帮助，通常都能找到解决办法。

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]

Answer 3

这个问题虽然已经有点时间了，但我有一个我认为比之前的解决方案简单一点的方法。

主要的观察是，当调用 command.current 时，alembic 并不是使用 Python 自带的 print 函数，而是用配置对象上的 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

Answer 4

我建议使用stdout的Config()对象参数（可以在这里查看）来把系统的输出（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)

Answer 5

你可以使用 MigrationContext 来 获取当前的版本：

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 的说法）。所以你可以用任何你想要的方式写代码，只要最终实现这个目的就可以了。