Async DRF类ModelView异步实现示例及单元测试方案咨询

一、异步ModelViewSet实现示例 #

类视图异步化的核心是将__call__方法声明为异步，同时将视图内的处理逻辑（包括list、retrieve等标准动作）改为异步实现，所有同步ORM操作需要用asgiref.sync.sync_to_async包裹避免阻塞事件循环，参考实现如下：

from rest_framework.viewsets import ModelViewSet
from rest_framework.response import Response
from rest_framework import status
from asgiref.sync import sync_to_async
from .models import Book
from .serializers import BookSerializer

class AsyncBookViewSet(ModelViewSet):
    queryset = Book.objects.all()
    serializer_class = BookSerializer

    # 重写异步__call__方法
    async def __call__(self, request, *args, **kwargs):
        self.args = args
        self.kwargs = kwargs
        request = self.initialize_request(request, *args, **kwargs)
        self.request = request
        self.headers = self.default_response_headers

        try:
            await sync_to_async(self.initial)(request, *args, **kwargs)
            # 获取对应的动作方法
            if request.method.lower() in self.http_method_names:
                handler = getattr(self, request.method.lower(), self.http_method_not_allowed)
            else:
                handler = self.http_method_not_allowed
            # 异步调用处理方法
            response = await handler(request, *args, **kwargs)
        except Exception as exc:
            response = await sync_to_async(self.handle_exception)(exc)
        
        response = self.finalize_response(request, response, *args, **kwargs)
        return response

    # 异步实现list动作
    async def list(self, request, *args, **kwargs):
        queryset = await sync_to_async(list)(self.filter_queryset(self.get_queryset()))
        page = await sync_to_async(self.paginate_queryset)(queryset)
        if page is not None:
            serializer = self.get_serializer(page, many=True)
            return self.get_paginated_response(serializer.data)
        serializer = self.get_serializer(queryset, many=True)
        return Response(serializer.data)

    # 其他动作retrieve/create/update/partial_update/destroy同理，按需异步实现
    async def retrieve(self, request, *args, **kwargs):
        instance = await sync_to_async(self.get_object)()
        serializer = self.get_serializer(instance)
        return Response(serializer.data)

实现注意事项：

• 内置的同步方法比如initial、handle_exception、filter_queryset、paginate_queryset、get_object都需要用sync_to_async包裹后异步调用
• 直接返回ORM查询集的操作需要用sync_to_async(list)(queryset)触发实际的数据库查询
• DRF本身的序列化逻辑不涉及IO的话可以不用异步包裹，保持同步调用即可

二、异步类视图单元测试示例 #

DRF 3.14+ 提供了AsyncAPIClient和AsyncAPITestCase支持异步视图测试，异步函数打桩可以使用Python标准库的AsyncMock实现。

pytest风格测试示例 #

import pytest
from rest_framework.test import AsyncAPIClient
from unittest.mock import AsyncMock

@pytest.mark.django_db
@pytest.mark.asyncio
async def test_async_book_list():
    # 初始化异步测试客户端
    client = AsyncAPIClient()
    # 异步调用接口
    response = await client.get("/api/books/")
    # 断言响应
    assert response.status_code == 200
    assert len(response.json()) == 0

@pytest.mark.django_db
@pytest.mark.asyncio
async def test_async_book_retrieve_mocked():
    # 异步方法打桩示例
    from .views import AsyncBookViewSet
    AsyncBookViewSet.retrieve = AsyncMock(return_value=Response({"id":1, "name":"测试书籍"}))
    client = AsyncAPIClient()
    response = await client.get("/api/books/1/")
    assert response.status_code == 200
    assert response.json()["name"] == "测试书籍"

unittest风格测试示例 #

from rest_framework.test import AsyncAPITestCase
from unittest.mock import AsyncMock
from asgiref.sync import sync_to_async
from .models import Book

class AsyncBookViewTest(AsyncAPITestCase):
    async def test_create_book(self):
        data = {"name": "测试书籍", "author": "测试作者"}
        response = await self.client.post("/api/books/", data=data, format="json")
        self.assertEqual(response.status_code, 201)
        # 验证数据库写入
        count = await sync_to_async(Book.objects.count)()
        self.assertEqual(count, 1)

测试注意事项：

• 测试异步视图必须使用异步客户端，不能用默认的同步APIClient
• 涉及数据库查询的断言操作也需要用sync_to_async包裹
• 异步方法的打桩必须使用AsyncMock，不能用普通的Mock或者MagicMock

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