从应用程序代码中使用 Alembic API

我使用 SQLite 作为基于 PySide 的桌面应用程序的应用程序文件格式(请参阅 给你了解为什么要这样做)。也就是说，当用户使用我的应用程序时，他们的数据被保存在他们机器上的单个数据库文件中。我正在使用 SQLAlchemyORM 与数据库进行通信。

当我发布应用程序的新版本时，我可能会修改数据库模式。我不希望用户在我每次更改模式时都必须丢弃他们的数据，所以我需要将他们的数据库迁移到最新的格式。此外，我还创建了许多临时数据库，以保存数据的子集，以便与一些外部进程一起使用。我想创建这些数据库与 alembic，使他们与正确的版本标记。

我有几个问题:

  • 有没有办法从 Python 代码中调用 alembic？我认为将 Popen用于纯 Python 模块很奇怪，但文档只是从命令行使用 alembic。主要是，我需要更改数据库位置到用户的数据库所在的位置。

  • 如果不可能，我可以从命令行指定一个新的数据库位置而不编辑。Ini 文件？这将使得通过 Popen呼叫 Alembic 变得不那么重要。

  • 我看到 alembic 将其版本信息保存在一个名为 alembic_version的简单表中，其中一列名为 version_num，一行指定版本。当我创建新数据库时，是否可以在模式中添加一个 alembic_version表并用最新版本填充它，这样就没有开销了？这是一个好主意吗; 我应该仅仅使用 alembic 来创建所有的数据库吗？

我有阿伦比克工作伟大的单一数据库，我用来开发与我的项目的目录。我想使用 alembic 方便地迁移和创建任意位置的数据库，最好是通过某种 Python API，而不是命令行。这个应用程序也会被 cx _ Frost 冻结，以防有什么不同。

谢谢！

32712 次浏览
小开

这是一个非常宽泛的问题，实际执行你的想法将取决于你，但它是可能的。

您可以在不使用命令的情况下从 Python 代码中调用 Alembic，因为它也是在 Python 中实现的！您只需要重新创建命令在幕后的操作。

无可否认，这些文档的状态不是很好，因为它们仍然是库的早期版本，但是通过一些挖掘，你会发现以下内容:

  1. 创建一个 配置
  2. 使用 Config 创建 ScriptDirectory 脚本目录
  3. 使用 Config 和 ScriptDirectory 创建 环境背景
  4. 使用 Environment Context 创建 迁移背景
  5. 大多数命令使用来自 Config 和 MigationContext 的一些方法的组合

我已经编写了一个扩展来提供对 Flask-SQLAlchemy 数据库的程序化 Alembic 访问。该实现与 Flask 和 Flask-SQLAlchemy 紧密相关，但它应该是一个很好的起点。参见 Flask-Alembic。

关于如何创建新数据库的最后一点，可以使用 Alembic 创建表，也可以使用 metadata.create_all()然后 alembic stamp head(或等效的 python 代码)。我建议始终使用迁移路径来创建表，而忽略原始的 metadata.create_all()。

我没有使用 cx _ zen 的经验，但是只要发行版中包含迁移，并且代码中到该目录的路径是正确的，那么就没有问题。

小开

以下是我将我的软件连接到 alembic后学到的东西:

有没有办法从 Python 代码中调用 alembic？

是的。在撰写本文时，alembic 的主要入口点是 alembic.config.main，因此您可以导入它并自己调用它，例如:

import alembic.config
alembicArgs = [
'--raiseerr',
'upgrade', 'head',
]
alembic.config.main(argv=alembicArgs)

请注意，alembic 在工作目录中寻找迁移(即 os.getcwd ())。我已经在调用 alembic 之前使用 os.chdir(migration_directory)处理了这个问题，但是可能有更好的解决方案。

我是否可以在不编辑. ini 文件的情况下从命令行指定一个新的数据库位置？

是的。关键在于 -x命令行参数。来自 alembic -h(令人惊讶的是，我在文档中找不到命令行参数引用) :

optional arguments:
-x X                  Additional arguments consumed by custom env.py
scripts, e.g. -x setting1=somesetting -x
setting2=somesetting

因此，您可以创建自己的参数，例如 dbPath，然后在 env.py中拦截它:

alembic -x dbPath=/path/to/sqlite.db upgrade head

例如 env.py:

def run_migrations_online():
# get the alembic section of the config file
ini_section = config.get_section(config.config_ini_section)


# if a database path was provided, override the one in alembic.ini
db_path = context.get_x_argument(as_dictionary=True).get('dbPath')
if db_path:
ini_section['sqlalchemy.url'] = db_path


# establish a connectable object as normal
connectable = engine_from_config(
ini_section,
prefix='sqlalchemy.',
poolclass=pool.NullPool)


# etc

当然，也可以在 alembic.config.main中使用 argv提供-x 参数。

我同意 @ 大卫主义关于使用迁移的观点，而不是 metadata.create_all():)

小开

下面是一个纯编程示例，说明如何以编程方式配置和调用 alembic 命令。

目录 setup (为了更容易读取代码)

.                         # root dir
|- alembic/               # directory with migrations
|- tests/diy_alembic.py   # example script
|- alembic.ini            # ini file

这是 diy _ alembic. py

import os
import argparse
from alembic.config import Config
from alembic import command
import inspect


def alembic_set_stamp_head(user_parameter):
# set the paths values
this_file_directory = os.path.dirname(os.path.abspath(inspect.stack()[0][1]))
root_directory      = os.path.join(this_file_directory, '..')
alembic_directory   = os.path.join(root_directory, 'alembic')
ini_path            = os.path.join(root_directory, 'alembic.ini')


# create Alembic config and feed it with paths
config = Config(ini_path)
config.set_main_option('script_location', alembic_directory)
config.cmd_opts = argparse.Namespace()   # arguments stub


# If it is required to pass -x parameters to alembic
x_arg = 'user_parameter=' + user_parameter
if not hasattr(config.cmd_opts, 'x'):
if x_arg is not None:
setattr(config.cmd_opts, 'x', [])
if isinstance(x_arg, list) or isinstance(x_arg, tuple):
for x in x_arg:
config.cmd_opts.x.append(x)
else:
config.cmd_opts.x.append(x_arg)
else:
setattr(config.cmd_opts, 'x', None)


#prepare and run the command
revision = 'head'
sql = False
tag = None
command.stamp(config, revision, sql=sql, tag=tag)


#upgrade command
command.upgrade(config, revision, sql=sql, tag=tag)

该代码或多或少是从 这个烧瓶-Alembic 文件的一个削减。这是查看其他命令使用情况和详细信息的好地方。

为什么是这个解决方案？-在运行自动化测试时，它是为了创建一个模糊的标记、升级和降级而编写的。

  • Chdir (偏移 _ 目录)干扰了一些测试。
  • 我们希望有一个数据库创建和操作的来源。“如果我们使用 alembic 来创建和管理数据库，那么测试也将使用 alembic 而不是 metadata.create _ all () shell”。
  • 即使上面的代码超过4行，如果以这种方式驱动，Alembic 显示出自己是一个很好的可控野兽。
小开

对于任何其他尝试使用 SQLAlchemy 获得类似苍蝇飞行的结果的人来说，这对我很有用:

将 偏移添加到项目中:

from flask_alembic import Alembic


def migrate(app):
alembic = Alembic()
alembic.init_app(app)
with app.app_context():
alembic.upgrade()

在初始化数据库之后，在应用程序启动时调用它

application = Flask(__name__)
db = SQLAlchemy()
db.init_app(application)
migration.migrate(application)

然后你只需要完成剩下的标准步骤:

将项目初始化为 alembic

alembic init alembic

更新 env.py:

from models import MyModel
target_metadata = [MyModel.Base.metadata]

更新 alembic.ini

sqlalchemy.url = postgresql://postgres:postgres@localhost:5432/my_db

假设您的 SQLAlchemy 模型已经定义好了，那么您现在就可以自动生成脚本了:

alembic revision --autogenerate -m "descriptive migration message"

如果出现无法在 env.py 中导入模型的错误，可以在您的 fo 补丁终端运行以下命令

export PYTHONPATH=/path/to/your/project

最后，我的迁移脚本是在 alembic/version 目录中生成的，我必须将它们复制到迁移目录中，以便 alembic 能够获取它们。

├── alembic
│   ├── env.py
│   ├── README
│   ├── script.py.mako
│   └── versions
│       ├── a5402f383da8_01_init.py  # generated here...
│       └── __pycache__
├── alembic.ini
├── migrations
│   ├── a5402f383da8_01_init.py  # manually copied here
│   └── script.py.mako

我可能有一些配置错误，但它现在工作。

小开

请参阅 alembic.opers.base 的文档。操作:

    from alembic.runtime.migration import MigrationContext
from alembic.operations import Operations


conn = myengine.connect()
ctx = MigrationContext.configure(conn)
op = Operations(ctx)


op.alter_column("t", "c", nullable=True)
小开

我没有使用 Flask，所以我不能使用已经推荐的 Flask-Alembic 图书馆。相反，经过一番修改之后，我编写了下面这个简短的函数来运行所有适用的迁移。我把所有与 Alembic 相关的文件都保存在一个称为偏移的子模块(文件夹)下。实际上我把 alembic.ini和 env.py放在一起，这可能有点不正统。下面是我的 alembic.ini文件中的一个片段，可以对其进行调整:

[alembic]
script_location = .

然后，我在同一目录中添加了以下文件，并将其命名为 run.py。但是无论你在哪里保存你的脚本，你所需要做的就是修改下面的代码指向正确的路径:

from alembic.command import upgrade
from alembic.config import Config
import os




def run_sql_migrations():
# retrieves the directory that *this* file is in
migrations_dir = os.path.dirname(os.path.realpath(__file__))
# this assumes the alembic.ini is also contained in this same directory
config_file = os.path.join(migrations_dir, "alembic.ini")


config = Config(file_=config_file)
config.set_main_option("script_location", migrations_dir)


# upgrade the database to the latest revision
upgrade(config, "head")

然后有了 run.py文件，我就可以在主代码中完成这个操作:

from mymodule.migrations.run import run_sql_migrations




run_sql_migrations()
小开

如果您查看来自 alembic 文档的 指令 API页面，您将看到一个如何直接从 Python 应用程序运行 CLI 命令的示例。不需要通过 CLI 代码。

运行 alembic.config.main的缺点是执行的 env.py脚本可能不是您想要的。例如，它将修改您的日志配置。

另一种非常简单的方法是使用上面链接的“命令 API”。例如，这里有一个小的 helper 函数，我最后写道:

from alembic.config import Config
from alembic import command


def run_migrations(script_location: str, dsn: str) -> None:
LOG.info('Running DB migrations in %r on %r', script_location, dsn)
alembic_cfg = Config()
alembic_cfg.set_main_option('script_location', script_location)
alembic_cfg.set_main_option('sqlalchemy.url', dsn)
command.upgrade(alembic_cfg, 'head')

我在这里使用 set_main_option方法是为了能够在需要的时候在不同的数据库上运行迁移。所以我可以简单地称之为:

run_migrations('/path/to/migrations', 'postgresql:///my_database')

从哪里获得这两个值(path 和 DSN)取决于您自己。但这似乎与你想要实现的目标非常接近。命令 API 还有 盖章()方法，它允许您将给定的 DB 标记为特定版本。上面的示例可以很容易地调用为。

小开

不算是答案，但是我很难接受，所以我想和大家分享一下:

如何使用 alembic.command.update 以编程方式传递 x _ 参数:

class CmdOpts:
x = {"data=true"}

在这里 data = true 是我在命令行中作为 x _ reference 传递的值

    alembic_config = AlembicConfig(ini_location)
setattr(alembic_config, "cmd_opts", CmdOpts())
alembic_config.cmd_opts.x = {"data": True}
小开

Alembic 将其所有命令公开为 alembic.command下的可导入调用。

Https://alembic.sqlalchemy.org/en/latest/api/commands.html

我编写了这个包装器，只是为了能够通过 python 代码设置自定义日志记录。

import logging


import alembic.command
import alembic.config


from somewhere import config_logging




def run():
config_logging()


log = logging.getLogger(__name__)


if len(sys.argv) < 3:
log.error("command must be specified")
exit(1)


else:
command_name = sys.argv[2]


try:
command = getattr(alembic.command, name)


except AttributeError:
log.error(f"{name} is not a valid alembic command")
exit(2)


config = alembic.config.Config()
config.set_main_option("script_location", "path/to/alembic")
config.set_main_option("sqlalchemy.url", "postgres://...")


command(config, *sys.argv[3:])

提交答案