pytest alembic用异步迁移初始化数据库

Question

现有帖子没有为我提供有用的答案。

我正在尝试使用pytest（db是asyncpg的postgres）来运行异步数据库测试，我想使用我的数据库初始化我的数据库Alembic迁移，以便我可以在此期间验证它们是否正常工作。

我的第一个尝试是：

@pytest.fixture(scope="session")
async def tables():
    """Initialize a database before the tests, and then tear it down again"""
    alembic_config: config.Config = config.Config('alembic.ini')
    command.upgrade(alembic_config, "head")
    yield
    command.downgrade(alembic_config, "base")

它实际上根本没有做任何事情（迁移从未应用于数据库，未创建表）。

Alembic的文档和amp; Pytest-alembic的文档说，异步迁移应通过配置您的 env> env 这样运行：

async def run_migrations_online() -> None:
    """Run migrations in 'online' mode.

    In this scenario we need to create an Engine
    and associate a connection with the context.

    """
    connectable = engine

    async with connectable.connect() as connection:
        await connection.run_sync(do_run_migrations)

    await connectable.dispose()

asyncio.run(run_migrations_online())

但这并不能解决问题（但是它确实适用于Pytest以外的生产迁移>）。

我跌倒在一个名为 pytest-alembic 的库上，该库为此提供了一些内置的测试。

运行 pytest -test-alembic 时，我会得到以下例外：

将未来附属于其他循环

上面有几个评论 pytest-asyncio 的github存储库表明，以下固定装置可能会解决它：

@pytest.fixture(scope="session")
def event_loop() -> Generator:
    loop = asyncio.get_event_loop_policy().new_event_loop()
    yield loop
    loop.close()

但是它没有（同样的例外）。

接下来，我尝试运行升级手动测试，使用

async def test_migrations(alembic_runner):
    alembic_runner.migrate_up_to("revision_tag_here")

：

alembic_runner.migrate_up_to（“ revision_tag_here”）

venv/lib/python3.9/site-packages/pytest_alembic/runner.py：264：在run_connection_task中 返回asyncio.run（run（Engine））

runtimeerror：asyncio.run（）无法从运行事件循环

调用

，但是这是内部呼叫 by pytest-alembic ，我不会我自己调用 asyncio.run（）我自己，所以我无法应用任何在线修复程序（ try-catch 以检查是否有一个现有的事件循环要使用等）。我敢肯定，这与我自己的 asyncio.run（）在 alembic env 中定义 - 或只是提出异常 - 该行实际上是从未执行的。

最后，我还尝试了 nest-asyncio.apply（），它永远悬挂。

还有一些博客文章建议使用此固定装置来初始化数据库表进行测试：

    async with engine.begin() as connection:
        await connection.run_sync(Base.metadata.create_all)

works 以创建一个数据库以进行测试，但这不是通过迁移运行的t帮助我的案子。

我觉得我已经尝试了所有的东西＆amp;访问了每个文档页面，但到目前为止我还没有运气。运行异步迁移测试肯定这不是很难吗？

如果需要任何额外的信息，我很高兴提供它。

Answer 1

以下 env.py 轻松地运行了这个问题

- 这里的主要思想是可以同步运行迁移

import asyncio
from logging.config import fileConfig

from alembic import context
from sqlalchemy import engine_from_config
from sqlalchemy import pool
from sqlalchemy.ext.asyncio import AsyncEngine

config = context.config

if config.config_file_name is not None:
    fileConfig(config.config_file_name)

target_metadata = mymodel.Base.metadata


def run_migrations_online():
    connectable = context.config.attributes.get("connection", None)
    if connectable is None:
        connectable = AsyncEngine(
            engine_from_config(
                context.config.get_section(context.config.config_ini_section),
                prefix="sqlalchemy.",
                poolclass=pool.NullPool,
                future=True
            )
        )

    if isinstance(connectable, AsyncEngine):
        asyncio.run(run_async_migrations(connectable))
    else:
        do_run_migrations(connectable)


async def run_async_migrations(connectable):
    async with connectable.connect() as connection:
        await connection.run_sync(do_run_migrations)
    await connectable.dispose()


def do_run_migrations(connection):
    context.configure(
        connection=connection,
        target_metadata=target_metadata,
        compare_type=True,
    )
    with context.begin_transaction():
        context.run_migrations()


run_migrations_online()

我通过
init_db.py

from alembic import command
from alembic.config import Config
from sqlalchemy.ext.asyncio import create_async_engine

__config_path__ = "/path/to/alembic.ini"
__migration_path__ = "/path/to/folder/with/env.py"

cfg = Config(__config_path__)
cfg.set_main_option("script_location", __migration_path__)


async def migrate_db(conn_url: str):
    async_engine = create_async_engine(conn_url, echo=True)
    async with async_engine.begin() as conn:
        await conn.run_sync(__execute_upgrade)


def __execute_upgrade(connection):
    cfg.attributes["connection"] = connection
    command.upgrade(cfg, "head")

然后您的pytest夹具看起来像这样
conftest.py

...

@pytest_asyncio.fixture(autouse=True)
async def migrate():
    await migrate_db(conn_url)
    yield

...

注意：我不将迁移的固定装置范围范围示为测试会话，我倾向于每次测试后掉落和迁移。

Answer 2

现在有一个Alembic现在提供的异步模板，请参见。

您可以使用以下内容生成异步 env.py ：

alembic init -t async <script_directory_here>

但请继续阅读到 https://alembic.sqlalchemy.org/en/latest/latest/latest/cookbook.html#programmat-programmatic-api-pogrammat--api-api-use-connection-connection-nection-with-with-with-with-with-with-with-with-with-with-with-ashncio 建议将 run_migrations_online 更改为以下内容：

def run_migrations_online():
    """Run migrations in 'online' mode."""

    connectable = config.attributes.get("connection", None)

    if connectable is None:
        asyncio.run(run_async_migrations())
    else:
        do_run_migrations(connectable)