You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

是否可在Django TestCase中同时使用async与数据库调用?

问题原因

psycopg2.InterfaceError: connection already closed 报错出现在 Django TestCase 异步测试场景下,核心逻辑是:TestCase 依赖单线程绑定的事务实现测试完成后快速回滚,默认数据库连接和运行测试的同步线程强绑定。异步方法中跨线程调用数据库操作时,新线程无法拿到当前测试事务绑定的有效连接,就会触发连接已关闭报错。

基于 TestCase 的解决方案(无性能损耗)

只需要调整数据库操作的封装规则,全程保留 TestCase 的事务回滚能力,不会出现 TransactionTestCase 重建表的性能问题:

  1. 所有数据库读写逻辑必须全部放到 database_sync_to_async 包裹的同步函数内部,禁止在异步上下文中提前构建 QuerySet 或调用数据库相关方法
  2. 给 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

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.10.01 23:09:03