Django/DRF异步测试触发SynchronousOnlyOperation错误，求排查

问题描述 #

用pytest完成Django/DRF项目的同步测试后，因速度较慢尝试用pytest-asyncio实现异步测试，触发以下错误：

django.core.exceptions.SynchronousOnlyOperation: You cannot call this from an async context - use a thread or sync_to_async

已尝试用asgiref.sync_to_async包装操作，但问题仍未解决。测试模块用于图片下载功能，包含授权/未授权客户端、测试用户等夹具，异步测试用例及同步正常版本如下：

异步测试代码（报错版本） #

@pytest_asyncio.fixture(loop_scope="module")
async def current_loop():
    yield asyncio.get_running_loop()


@pytest.mark.django_db
@pytest.fixture
async def user() -> User:
    """ Creating test user """

    yield User.objects.create_superuser(username='test.user', password='user.password')


@pytest.mark.django_db
@pytest_asyncio.fixture
async def auth_client(user: User) -> Iterator[APIClient]:
    """ Authorised client session """

    client = await sync_to_async(APIClient())()
    await sync_to_async(client.force_login(user))()
    yield client


@pytest.mark.django_db
@pytest_asyncio.fixture
async def non_auth_client(user: User) -> Iterator[APIClient]:
    """ Non-authorised client """

    client = await sync_to_async(APIClient())()
    yield client


@pytest_asyncio.fixture
async def model(current_loop):
    model_case = await sync_to_async(document_models.LicensesAndCertificatesImages)()
    yield model_case
    # teardown
    for model_object in model_case.objects.all():
        model_object.delete()


@pytest_asyncio.fixture
async def post_url():
    post_url_case = reverse('admin_site:documents_licenses-upload')
    yield post_url_case


@pytest_asyncio.fixture
async def get_list_url():
    get_list_url_case = reverse('admin_site:documents_licenses-list')
    return get_list_url_case


@pytest.mark.django_db
@pytest_asyncio.fixture
async def prepare_models_objects(user, model):
    """ Filling DB with model objects for GET-requests testing """

    uploaded_by = await sync_to_async(get_user_model().objects.first())()

    for i in range(5):
        file = SimpleUploadedFile(
            name=f'file_{i}.png',
            content=b'correct_file_content',
            content_type='image/png',
        )
        await model.objects.create(document=file, uploaded_by=uploaded_by)
    yield


""" Tests """


@pytest.mark.django_db
async def test_get_list(prepare_models_objects, model, auth_client, get_list_url):
    """ Check if GET-request receives License and Sertificates list """

    response: Response = await sync_to_async(auth_client.get(path=get_list_url))()

    assert response.status_code == 200
    assert len(response.data) == 5
    assert list(response.data[0].keys()) == ['id', 'document']

同步正常版本 #

@pytest.mark.django_db
def test_get_list(prepare_models_objects, model, auth_client, get_list_url):
    response: Response = auth_client.get(path=get_list_url)

    assert response.status_code == 200
    assert len(response.data) == 5
    assert list(response.data[0].keys()) == ['id', 'document']

错误原因分析 #

1. 异步fixture定义不规范：user fixture是异步函数，但未添加@pytest_asyncio.fixture装饰器，且内部直接调用同步ORM方法User.objects.create_superuser，未用sync_to_async包装。
2. APIClient使用方式错误：sync_to_async(APIClient())()仅同步创建了APIClient实例，但后续调用client.get时，内部的同步逻辑（如请求处理、ORM查询）仍运行在异步上下文，未被正确隔离到同步线程。
3. ORM操作未正确包装：prepare_models_objects中的model.objects.create是同步ORM方法，直接在异步函数中调用；model fixture的teardown部分，model_case.objects.all()和model_object.delete()都是同步操作，均未做异步兼容处理。

修复方案 #

1. 规范异步fixture定义 #

所有异步fixture必须添加@pytest_asyncio.fixture装饰器，且同步ORM操作必须用sync_to_async完整包装。

2. 正确处理APIClient调用 #

两种可选方案：

• 方案一：用sync_to_async完整包装APIClient的操作逻辑，确保整个请求流程在同步线程执行。
• 方案二：如果DRF版本≥3.14，直接使用DRF提供的AsyncAPIClient，它原生支持异步请求。

3. 修复teardown中的同步操作 #

将model fixture的清理逻辑用sync_to_async包装，或者封装成同步函数后通过sync_to_async调用。

4. 完整包装所有ORM同步方法 #

所有Django ORM的同步方法（create、filter、delete等）都需要通过sync_to_async调用，不能直接在异步函数中执行。

修复后的示例代码 #

import asyncio
from asgiref.sync import sync_to_async
from django.contrib.auth import get_user_model
from django.core.files.uploadedfile import SimpleUploadedFile
from django.urls import reverse
import pytest
import pytest_asyncio
from rest_framework.test import APIClient, Response
from your_app import document_models  # 替换为实际的模型导入路径

User = get_user_model()


@pytest_asyncio.fixture(loop_scope="module")
async def current_loop():
    yield asyncio.get_running_loop()


@pytest.mark.django_db
@pytest_asyncio.fixture
async def user() -> User:
    """ Creating test user """
    # 用sync_to_async包装同步ORM创建方法
    user = await sync_to_async(User.objects.create_superuser)(
        username='test.user', password='user.password'
    )
    yield user
    # 清理用户
    await sync_to_async(user.delete)()


@pytest.mark.django_db
@pytest_asyncio.fixture
async def auth_client(user: User) -> Iterator[APIClient]:
    """ Authorised client session """
    # 同步创建客户端并登录，整个逻辑包装在sync_to_async中
    async def create_auth_client():
        client = APIClient()
        client.force_login(user)
        return client

    client = await sync_to_async(create_auth_client)()
    yield client


@pytest.mark.django_db
@pytest_asyncio.fixture
async def non_auth_client() -> Iterator[APIClient]:
    """ Non-authorised client """
    client = await sync_to_async(APIClient)()
    yield client


@pytest_asyncio.fixture
async def model():
    model_case = document_models.LicensesAndCertificatesImages
    yield model_case
    # 用sync_to_async包装清理逻辑
    await sync_to_async(model_case.objects.all().delete)()


@pytest_asyncio.fixture
async def post_url():
    post_url_case = reverse('admin_site:documents_licenses-upload')
    yield post_url_case


@pytest_asyncio.fixture
async def get_list_url():
    get_list_url_case = reverse('admin_site:documents_licenses-list')
    return get_list_url_case


@pytest.mark.django_db
@pytest_asyncio.fixture
async def prepare_models_objects(user, model):
    """ Filling DB with model objects for GET-requests testing """
    uploaded_by = await sync_to_async(get_user_model().objects.first)()

    for i in range(5):
        file = SimpleUploadedFile(
            name=f'file_{i}.png',
            content=b'correct_file_content',
            content_type='image/png',
        )
        # 包装model.objects.create为异步方法
        await sync_to_async(model.objects.create)(
            document=file, uploaded_by=uploaded_by
        )
    yield


""" Tests """


@pytest.mark.django_db
async def test_get_list(prepare_models_objects, model, auth_client, get_list_url):
    """ Check if GET-request receives License and Sertificates list """
    # 用sync_to_async包装整个get请求调用
    response: Response = await sync_to_async(auth_client.get)(path=get_list_url)

    assert response.status_code == 200
    assert len(response.data) == 5
    assert list(response.data[0].keys()) == ['id', 'document']

额外建议 #

• 如果你的DRF版本≥3.14，推荐使用AsyncAPIClient替代APIClient，它原生支持异步请求，无需手动用sync_to_async包装：

  from rest_framework.test import AsyncAPIClient
  
  @pytest_asyncio.fixture
  async def auth_client(user: User):
      client = AsyncAPIClient()
      await client.force_login(user)
      yield client
  

• 对于批量ORM操作，可考虑用sync_to_async包装整个批量逻辑，减少线程切换开销。

内容的提问来源于stack exchange，提问作者TiimeIrre