使用 FastAPI 和 ormar 切换到测试数据库

Question

我正在使用 FastAPI 尝试 ormar 并努力进行测试。因为我在我的大部分项目中使用 Django，所以我试图在开发和测试之间分离我的数据库。但我正在努力做到这一点。

目标基本上是，development/production 使用 PostgreSQL，但在运行 pytest 时切换到 SQLite。并在测试完成后从 SQLite 数据库中删除所有数据。（想象 Django+Docker 开发环境）

我尝试了 ormar 文档中的以下内容，但没有成功。

TEST_DB_URL = "sqlite:///../db.sqlite"

@pytest.fixture(autouse=True, scope="module")
def create_test_database():
    engine = sqlalchemy.create_engine(TEST_DB_URL)
    metadata.create_all(engine)
    yield
    metadata.drop_all(engine)

完成上述操作后，当我尝试在测试中对数据库执行某些操作时（使用 pytest），出现以下错误。

====================================================================================== FAILURES =======================================================================================
_______________________________________________________________________________ test_user_registration ________________________________________________________________________________

db = <function db.<locals>.wrapper at 0x7fb7d30c9820>

    @pytest.mark.asyncio
    async def test_user_registration(db):
>       users = await db(User.objects.all)

app/users/test_users.py:51: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
app/conftest.py:39: in wrapper
    return await func()
/usr/local/lib/python3.9/site-packages/ormar/queryset/queryset.py:1016: in all
    rows = await self.database.fetch_all(expr)
/usr/local/lib/python3.9/site-packages/databases/core.py:147: in fetch_all
    async with self.connection() as connection:
/usr/local/lib/python3.9/site-packages/databases/core.py:251: in __aenter__
    raise e
/usr/local/lib/python3.9/site-packages/databases/core.py:248: in __aenter__
    await self._connection.acquire()
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

self = <databases.backends.postgres.PostgresConnection object at 0x7fb7d319e910>

    async def acquire(self) -> None:
        assert self._connection is None, "Connection is already acquired"
>       assert self._database._pool is not None, "DatabaseBackend is not running"
E       AssertionError: DatabaseBackend is not running

/usr/local/lib/python3.9/site-packages/databases/backends/postgres.py:162: AssertionError
=============================================================================== short test summary info ===============================================================================
FAILED app/users/test_users.py::test_user_registration - AssertionError: DatabaseBackend is not running

我的数据库设置如下所示，通过常规 API 调用可以很好地保存数据等。

import os

import databases
from fastapi import FastAPI
import sqlalchemy

from app.resources.utils import get_models_path_list

models_path_list = get_models_path_list()


def get_db_uri(*, user, password, host, db):
    return f'postgresql://{user}:{password}@{host}:5432/{db}'


DB_URL = get_db_uri(
    user=os.environ.get('POSTGRES_USER'),
    password=os.environ.get('POSTGRES_PASSWORD'),
    host='db',  # docker-composeのservice名
    db=os.environ.get('POSTGRES_DB'),
)
database = databases.Database(DB_URL)
metadata = sqlalchemy.MetaData()


def setup_database(app: FastAPI):
    app.state.database = database

    @app.on_event("startup")
    async def startup() -> None:
        database_ = app.state.database
        if not database_.is_connected:
            await database_.connect()

    @app.on_event("shutdown")
    async def shutdown() -> None:
        database_ = app.state.database
        if database_.is_connected:
            await database_.disconnect()

Answer 1

我也在那里挣扎过。首先 this 讨论会对你有所帮助。症结在这里。您需要使用指向您的 sqlite 数据库的环境变量。我确实使用了 FastAPI 的 BaseSettings 并在那里添加了这个方法

def get_db_uri(self) -> str:
    # set the below in your environment file when running tests
    if self.TESTING:
        return "sqlite:///../db.sqlite"

    if self.PRODUCTION:
        return self._get_db_uri(
            user="root",
            passwd=self.POSTGRES_PASSWORD,
            host=self.AURORA_DB_URI,
            port=self.POSTGRES_PORT,
            db=self.POSTGRES_DB,
        )

    return self._get_db_uri(
        user=self.POSTGRES_USER,
        passwd=self.POSTGRES_PASSWORD,
        host=self.POSTGRES_HOST,
        port=self.POSTGRES_PORT,
        db=self.POSTGRES_DB,
    )

我缺少的另一件事是用于删除和创建表的固定装置。你已经在那里了。

TEST_DB_URL = "sqlite:///../db.sqlite"

@pytest.fixture(autouse=True, scope="module")
def create_test_database():
    engine = sqlalchemy.create_engine(TEST_DB_URL)
    metadata.create_all(engine)
    yield
    metadata.drop_all(engine)

您可以使用 pytest-env 在测试中设置环境变量。

Answer 2

对于那些想以类似项目的形式看到这个的人，我做了一个小项目来重现上面的答案。

https://github.com/joshua-hashimoto/fastapi-ormar

使用 FastAPI 和 ormar 切换到测试数据库

Switch to test database using FastAPI and ormar

python

pytest

fastapi

ormar