如何在异步 FastAPI 测试中使用 PostgreSQL 测试数据库？

Question

我正在开发一个异步 FastAPI 项目，我想在测试期间连接到数据库。来自 Django，我的本能是创建 pytest 固定装置来负责创建/删除测试数据库。但是，我找不到太多关于如何执行此操作的文档。我能找到的最完整的说明在本教程中，但它们对我不起作用，因为它们都是同步的。我对异步开发有点陌生，所以我在调整代码以异步工作时遇到了困难。这是我到目前为止所拥有的：

import pytest
from sqlalchemy.ext.asyncio import create_async_engine, session
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from sqlalchemy_utils import database_exists, create_database

from fastapi.testclient import TestClient

from app.core.db import get_session
from app.main import app


Base = declarative_base()


@pytest.fixture(scope="session")
def db_engine():
    default_db = (
        "postgresql+asyncpg://postgres:postgres@postgres:5432/postgres"
    )
    test_db = "postgresql+asyncpg://postgres:postgres@postgres:5432/test"
    engine = create_async_engine(default_db)
    if not database_exists(test_db):  # <- Getting error on this line
        create_database(test_db)

    Base.metadata.create_all(bind=engine)
    yield engine


@pytest.fixture(scope="function")
def db(db_engine):
    connection = db_engine.connect()

    # begin a non-ORM transaction
    connection.begin()

    # bind an individual Session to the connection
    Session = sessionmaker(bind=connection)
    db = Session()
    # db = Session(db_engine)

    yield db

    db.rollback()
    connection.close()


@pytest.fixture(scope="function")
def client(db):
    app.dependency_overrides[get_session] = lambda: db

    PREFIX = "/api/v1/my-endpoint"
    with TestClient(PREFIX, app) as c:
        yield c

这是我收到的错误：

E           sqlalchemy.exc.MissingGreenlet: greenlet_spawn has not been called; can't call await_() here. Was IO attempted in an unexpected place? (Background on this error at: https://sqlalche.me/e/14/xd2s)

/usr/local/lib/python3.9/site-packages/sqlalchemy/util/_concurrency_py3k.py:67: MissingGreenlet

知道我需要做什么来修复它吗？

Answer 1

您尝试将同步引擎与异步会话一起使用。尝试使用：

from sqlalchemy.ext.asyncio import AsyncSession
Session = sessionmaker(bind= connection, class_=AsyncSession)

https://docs.sqlalchemy.org/en/14/orm/extensions/asyncio.html