python - 以编程方式获取 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 文件检查它是否有一个格式化程序区域。这是我的 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 浏览器我可以看到数据库中的版本没有设置为无。

score 12 · Accepted Answer

您可以使用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）。所以你可以用任何你想要的方式编写代码，只要那是你最终要做的。

score 4 · Accepted Answer

我建议使用 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)

score 2 · Accepted Answer

这个问题很老，但我有一个解决方案，我认为它比迄今为止给出的解决方案要简单一些。

主要观察是，当command.current被调用时，alembic 不使用 Python 的内置print函数，而是使用print_stdoutconfig 对象上的方法。因此，要捕获输出只是重载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

score 1 · Accepted Answer

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]

score 1 · Accepted Answer

只是想投入我的 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 返回的版本是当前头还是完全有效。

score 1 · Accepted Answer

只是建立在詹姆斯芬内尔的答案上，在获取历史的情况下，它只会输出一行。如果您想要所有输出（作为列表），您可以执行以下操作：

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

score 0 · Accepted Answer

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

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

如果未找到版本号，此函数将返回 None。

score 0 · Accepted Answer

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

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

score 0 · Accepted Answer

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

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

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

9 回答 9

Related

Reference