时间戳

首页 > 解决方案 > 单元测试时如何从数据库异步获取记录？

python - 单元测试时如何从数据库异步获取记录？

问题描述

在这个单元测试中，我想检查设备是否已经创建并且到期日期是未来 7 天。

database.py

import databases
database = databases.Database(settings.sqlalchemy_database_uri)

单元测试：

from database.database import database

def test_successful_register_expiry_set_to_seven_days():
    response = client.post(
        "/register/",
        headers={},
        json={"device_id": "u1"},
    )
    assert response.status_code == 201
    query = device.select(whereclause=device.c.id == "u1")
    d = database.fetch_one(query)
    assert d.expires_at == datetime.utcnow().replace(microsecond=0) + timedelta(days=7)

因为d是协程对象，所以它失败并显示以下消息：

AttributeError: 'coroutine' object has no attribute 'expires_at'

而且我不能await在单元测试中使用。

d = await database.fetch_one(query)

请问我错过了什么？

标签： pythonsqlalchemyfastapi

解决方案

好吧，它永远不会被等待，您的代码在该信号量进入调度程序之前返回协程。

如果您使用的是异步驱动程序，则需要等待它。有什么解决方法吗？是的。

您可以使用asyncio.run(awaitable)在事件循环中运行协程。

import asyncio

d = asyncio.run(database.fetch_one(query))

如果您有一个当前正在运行的事件循环，您可能希望使用该事件循环。您可以通过 来实现asyncio.get_event_loop()，这将在运行循环中运行该函数。

import asyncio

asyncio.get_event_loop().run_until_complete(database.fetch_one(query))

您还可以使用@pytest.mark.asyncio装饰器（请参阅文档）。

@pytest.mark.asyncio
async def dummy():
    await some_awaitable()

推荐阅读