Vistas asincronas en DRF con Django 5

Cuando merece la pena ir asincrono

Una vista sincrona ocupa un worker entero hasta que termina, aunque pase la mayor parte del tiempo esperando una respuesta HTTP, una consulta a Redis o un fichero en S3. Si tienes 8 workers gunicorn y cada peticion tarda 2 segundos esperando a un servicio externo, tu API solo aguanta 4 req/s antes de empezar a encolar.

Una vista asincrona libera el bucle de eventos durante esa espera, asi que el mismo proceso puede atender cientos de conexiones simultaneas mientras los I/O ocurren. La regla es sencilla.

Async gana cuando la vista pasa la mayor parte del tiempo esperando I/O: APIs externas, llamadas a LLMs (OpenAI, Anthropic), webhooks salientes, agregadores que llaman a varios microservicios. Async no aporta nada (e incluso puede empeorar) en vistas que solo tocan tu base de datos local con queries cortas.

Configuracion: Django 5 y ASGI

Django 5 trae soporte estable de async views en DRF. Necesitas dos piezas.

# settings.py
INSTALLED_APPS = [
    "django.contrib.contenttypes",
    "django.contrib.auth",
    "rest_framework",
    "tienda",
]

# Sigues teniendo WSGI por compatibilidad, pero el server real sera ASGI:
WSGI_APPLICATION = "core.wsgi.application"
ASGI_APPLICATION = "core.asgi.application"

# core/asgi.py
import os
from django.core.asgi import get_asgi_application

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "core.settings")
application = get_asgi_application()

El servidor recomendado es uvicorn (basado en uvloop) detras de gunicorn con worker class uvicorn.workers.UvicornWorker.

gunicorn core.asgi:application \
  -w 4 \
  -k uvicorn.workers.UvicornWorker \
  --bind 0.0.0.0:8000

Si usas runserver, Django ya levanta automaticamente ASGI cuando detecta vistas async. Pero en produccion siempre uvicorn.

Primera APIView async

DRF detecta async def y lo trata correctamente. La unica diferencia frente a una APIView normal es la palabra clave async.

# views.py
from rest_framework.views import APIView
from rest_framework.response import Response
import asyncio

class PingAsyncView(APIView):
    async def get(self, request):
        await asyncio.sleep(0.5)  # Simula I/O
        return Response({"status": "ok", "tipo": "async"})

Internamente DRF envuelve la vista para llamar a get() con await. El permission/authentication checking sigue funcionando igual.

ORM asincrono

Desde Django 4.1 (y consolidado en 5) cada metodo del ORM tiene su variante a*.

from tienda.models import Producto

class ProductoDetailAsync(APIView):
    async def get(self, request, pk):
        # NO uses Producto.objects.get(pk=pk) en async (bloqueante)
        producto = await Producto.objects.aget(pk=pk)
        return Response({
            "id": producto.id,
            "nombre": producto.nombre,
            "precio": str(producto.precio),
        })

Equivalencias mas usadas:

• get() -> aget()
• create() -> acreate()
• save() -> asave()
• delete() -> adelete()
• filter() queryset es lazy, lo iteras con async for o lo materializas con [obj async for obj in queryset].
• count() -> acount()
• exists() -> aexists()

class ProductoListAsync(APIView):
    async def get(self, request):
        productos = [
            {"id": p.id, "nombre": p.nombre}
            async for p in Producto.objects.filter(disponible=True)
        ]
        return Response(productos)

El ORM async no soporta todavia algunas cosas avanzadas: select_for_update, transacciones complejas anidadas o algunos prefetch_related exoticos. Para esas, usa sync_to_async.

Llamadas HTTP externas en paralelo: httpx + asyncio.gather

El caso clasico donde async brilla. Imagina que tu vista tiene que consultar 3 microservicios y devolver un agregado.

# views.py
import httpx
import asyncio
from rest_framework.views import APIView
from rest_framework.response import Response

class DashboardView(APIView):
    async def get(self, request, user_id):
        async with httpx.AsyncClient(timeout=5.0) as client:
            usuario_t = client.get(f"http://users-svc/api/users/{user_id}/")
            pedidos_t = client.get(f"http://orders-svc/api/orders/?user={user_id}")
            recom_t   = client.get(f"http://reco-svc/api/recos/?user={user_id}")

            usuario_r, pedidos_r, recom_r = await asyncio.gather(
                usuario_t, pedidos_t, recom_t
            )

        return Response({
            "usuario":          usuario_r.json(),
            "pedidos":          pedidos_r.json(),
            "recomendaciones":  recom_r.json(),
        })

Las 3 llamadas se lanzan en paralelo y la vista termina cuando termina la mas lenta, no la suma. Si cada una tarda 200 ms, el total es ~200 ms y no 600 ms.

httpx es a requests lo que asyncio es a hilos. Misma API, pero soporta await. Instala con pip install httpx.

sync_to_async para codigo bloqueante

Si dentro de una vista async necesitas llamar a una libreria sincrona (por ejemplo boto3), envuelvela.

from asgiref.sync import sync_to_async
import boto3

s3 = boto3.client("s3")

def _upload(file_bytes, key):
    s3.put_object(Bucket="mi-bucket", Key=key, Body=file_bytes)

class UploadView(APIView):
    async def post(self, request):
        contenido = request.body
        await sync_to_async(_upload, thread_sensitive=False)(
            contenido, f"uploads/{request.user.id}.bin"
        )
        return Response({"ok": True})

thread_sensitive=False permite ejecutarlo en un pool de hilos sin bloquear al loop. Si lo dejas en True (por defecto), Django serializa esa llamada con el thread principal, lo que mata el paralelismo.

Streaming asincrono y SSE

Para respuestas largas o eventos en tiempo real, devuelve un StreamingHttpResponse con un generador asincrono.

from django.http import StreamingHttpResponse
import asyncio
import json

class ChatStreamView(APIView):
    async def get(self, request):
        async def event_stream():
            for i in range(10):
                yield f"data: {json.dumps({'chunk': i})}\n\n"
                await asyncio.sleep(0.5)

        return StreamingHttpResponse(
            event_stream(),
            content_type="text/event-stream",
        )

Patron muy util para integrar respuestas streaming de LLMs (OpenAI, Anthropic) directamente al cliente sin acumular toda la respuesta en memoria.

Errores comunes y como evitarlos

flowchart TD
    A[Vista async] --> B{Codigo bloqueante dentro?}
    B -->|si| C[sync_to_async + thread_sensitive=False]
    B -->|no| D[await directamente]
    D --> E{Multiples I/O independientes?}
    E -->|si| F[asyncio.gather]
    E -->|no| G[await secuencial]
    C --> H[Vista no bloquea el loop]
    F --> H
    G --> H

Tres errores recurrentes que rompen el rendimiento:

• Llamar a requests.get en una vista async. Bloquea el loop durante toda la latencia y anula la ventaja. Usa httpx.AsyncClient.
• Iterar un queryset con for obj in qs en vez de async for. Internamente Django ejecuta sync_to_async en modo thread-sensitive y cada iteracion bloquea.
• Usar time.sleep para esperar. Bloquea el thread completo. Usa siempre await asyncio.sleep.

Cuando NO usar async

Si tu vista solo hace una query a Postgres y devuelve un serializer, el coste de cambiar a async no compensa: el ORM async tiene un overhead pequeno, complica el codigo y muchos middlewares (cache, login_required) tienen rutas sync mas optimizadas.

Resume: async para I/O externo paralelo y streaming. Sincrono para CRUDs simples contra tu BBDD propia.