我正在处理一个异步FastAPI项目，我想在测试期间连接到数据库。来自Django，我的直觉是创建pytest fixture，负责创建/删除测试数据库。然而，我找不到太多关于如何做到这一点的文档。我能找到的最完整的说明在本教程中，但它们对我不起作用，因为它们都是同步的。我对异步开发有点陌生，所以我很难调整代码以使其异步工作。这就是我目前所拥有的：
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
你知道我该怎么办吗？
您尝试将sync引擎与async会话一起使用。尝试使用：
from sqlalchemy.ext.asyncio import AsyncSession
Session = sessionmaker(bind= connection, class_=AsyncSession)
https://docs.sqlalchemy.org/en/14/orm/extensions/asyncio.html