是否可在Django TestCase中同时使用async与数据库调用?
问题原因
psycopg2.InterfaceError: connection already closed报错出现在 DjangoTestCase异步测试场景下,核心逻辑是:TestCase依赖单线程绑定的事务实现测试完成后快速回滚,默认数据库连接和运行测试的同步线程强绑定。异步方法中跨线程调用数据库操作时,新线程无法拿到当前测试事务绑定的有效连接,就会触发连接已关闭报错。
基于 TestCase 的解决方案(无性能损耗)
只需要调整数据库操作的封装规则,全程保留 TestCase 的事务回滚能力,不会出现 TransactionTestCase 重建表的性能问题:
- 所有数据库读写逻辑必须全部放到
database_sync_to_async包裹的同步函数内部,禁止在异步上下文中提前构建 QuerySet 或调用数据库相关方法 - 给
database_sync_to_async显式指定thread_sensitive=True参数,强制所有同步数据库操作在同一个线程执行,复用测试事务绑定的连接
修改后代码示例
example.py
from django.contrib.auth import get_user_model from channels.db import database_sync_to_async async def func(): # 所有数据库逻辑封装到同步内部函数中 @database_sync_to_async(thread_sensitive=True) def query_user_exists(): return get_user_model().objects.filter(pk=1).exists() return await query_user_exists()
test_example.py
import django.test from .example import func class TestFunc(django.test.TestCase): async def test_func(self): res = await func() # Django 4.1+ 异步测试方法支持直接调用同步断言,无需手动包裹 sync_to_async self.assertFalse(res)
注意事项
- 如果你使用
asgiref.sync.sync_to_async封装数据库操作,默认thread_sensitive为False,必须手动指定为True才能生效 - 不要在异步上下文中提前初始化 QuerySet、模型实例的数据库关联属性,避免提前绑定无效的数据库连接
内容的提问来源于stack exchange,提问作者Dmitry
相关产品推荐
相关产品推荐

