Async Django, Channels, and Async ORM

Django got serious about async between 3.1 (2020) and 4.2 / 5.x. You can write async def views, call the ORM with await, and run WebSockets via Channels — but the migration story has rough edges worth knowing.

Async views

# views.py
import httpx
from django.http import JsonResponse

async def fetch_external(request):
    async with httpx.AsyncClient(timeout=5) as c:
        r = await c.get("https://api.example.com/data")
    return JsonResponse(r.json())

The view is async def; Django runs it through ASGI. WSGI servers (gunicorn + sync workers) can still serve async views by running the loop per-request — but you lose the concurrency benefit. Use ASGI (uvicorn / daphne / hypercorn) in production for async-heavy apps.

Async ORM (Django 4.1+)

from django.contrib.auth import get_user_model
User = get_user_model()

async def view(request):
    user = await User.objects.aget(id=request.user.id)            # async get
    orders = [o async for o in user.order_set.all()]              # async iteration
    count = await user.order_set.acount()                          # async count
    await user.asave()                                             # async save

Async methods are prefixed with a — aget, acreate, asave, adelete, acount, afirst, etc. Iteration over QuerySets uses async for.

Critical gotcha: the ORM is async at the API level, but the database driver is still synchronous underneath. Django wraps blocking driver calls in sync_to_async. You get the programming model of async but not the concurrency benefit — each query still blocks a thread.

For real async DB I/O, use psycopg (3.x) — its async support pairs with a properly-async driver. Django 5.x is moving toward this; check current state before relying on it for perf.

sync_to_async / async_to_sync

The bridge between sync and async Django code:

from asgiref.sync import sync_to_async, async_to_sync

async def my_async_view(request):
    # Call sync function from async context
    result = await sync_to_async(slow_sync_function)(arg)

# Call async function from sync context
def my_sync_view(request):
    result = async_to_sync(my_async_function)(arg)

sync_to_async runs the sync function in a thread pool. async_to_sync runs the async function in its own loop. Both add overhead — use sparingly.

Thread sensitivity: Django ORM and DB connections are thread-local. By default sync_to_async runs in a thread pool, so the connection might be on a different thread than expected. Use sync_to_async(fn, thread_sensitive=True) to force serial execution in the main thread (slower, but safer for ORM operations that depend on transaction state).

Channels — WebSockets and beyond

Django Channels extends Django to handle WebSockets, long-polling, server-sent events, and any non-HTTP protocol — anything ASGI can route.

# routing.py
from channels.routing import ProtocolTypeRouter, URLRouter
from django.urls import path
from chat.consumers import ChatConsumer

application = ProtocolTypeRouter({
    "http": django_asgi_app,
    "websocket": URLRouter([
        path("ws/chat/<room>/", ChatConsumer.as_asgi()),
    ]),
})

# consumers.py
from channels.generic.websocket import AsyncWebsocketConsumer
import json

class ChatConsumer(AsyncWebsocketConsumer):
    async def connect(self):
        self.room = self.scope["url_route"]["kwargs"]["room"]
        await self.channel_layer.group_add(self.room, self.channel_name)
        await self.accept()

    async def disconnect(self, code):
        await self.channel_layer.group_discard(self.room, self.channel_name)

    async def receive(self, text_data=None, bytes_data=None):
        data = json.loads(text_data)
        await self.channel_layer.group_send(self.room, {
            "type": "chat.message",
            "message": data["message"],
            "user": self.scope["user"].username if self.scope["user"].is_authenticated else "anon",
        })

    async def chat_message(self, event):
        await self.send(text_data=json.dumps(event))

Channel layers — Redis as the inter-process bus

Channels by default runs in a single process. To broadcast across many workers (typical production), use a channel layer — usually Redis.

# settings.py
CHANNEL_LAYERS = {
    "default": {
        "BACKEND": "channels_redis.core.RedisChannelLayer",
        "CONFIG": {"hosts": [("redis", 6379)]},
    },
}

channel_layer.group_send(...) publishes to Redis; every worker subscribed to the group receives it and forwards to its connected WebSockets. Standard fan-out pattern.

Gotcha: Redis is the SPOF and the bottleneck. At high WebSocket fan-out (10k+ connections), the channel layer cost dominates — consider running fewer workers with more connections each, or a Redis cluster.

Running async Django in production

ASGI server is the change:

# Old (sync only)
gunicorn myproject.wsgi:application

# New (async + sync)
uvicorn myproject.asgi:application --workers 4
# or with gunicorn driving uvicorn workers:
gunicorn myproject.asgi:application -k uvicorn.workers.UvicornWorker --workers 4

asgi.py is generated by django-admin startproject since 3.0. It uses get_asgi_application().

Each worker runs its own event loop. Use multiple workers to use multiple CPU cores (one event loop = one core). Within a worker, async I/O lets you handle many concurrent requests.

Middleware

Django middleware can be sync, async, or both. Modern pattern:

class MyMiddleware:
    sync_capable = True
    async_capable = True

    def __init__(self, get_response):
        self.get_response = get_response
        self._is_coroutine = iscoroutinefunction(get_response)

    def __call__(self, request):
        if self._is_coroutine:
            return self._async_call(request)
        return self._sync_call(request)

    async def _async_call(self, request):
        # do async pre-processing
        response = await self.get_response(request)
        # do async post-processing
        return response

    def _sync_call(self, request):
        ...

Most third-party middleware is sync-only; Django wraps it via sync_to_async. Performance suffers if you have a long middleware chain in an async app.

When does async Django win

• WebSockets / SSE / long-lived connections — Channels, no alternative in Django land.
• Heavy outbound HTTP / API aggregation — async http calls within a request actually parallel.
• Many slow external dependencies per request — asyncio.gather over awaits beats sync sequential.

When async Django mostly doesn’t pay off (yet):

• DB-bound services — async ORM thread-wraps the sync driver; no real concurrency gain.
• Existing big sync codebases — sync_to_async tax everywhere; not worth it for a marginal win.

When to use FastAPI instead

For greenfield async-first services with no Django admin / ORM / forms requirement, FastAPI is usually faster to build and runs natively async end-to-end (Starlette + asyncpg/SQLAlchemy 2.0 async). Django’s strength is everything around the ORM and admin; if you don’t need those, FastAPI is leaner.

Common gotchas

• Calling sync ORM from async view. User.objects.get(...) (no a) in an async view raises SynchronousOnlyOperation. Use aget.
• Querysets in async iteration. for u in User.objects.all() is sync; use async for u in User.objects.all().
• Middleware order matters more. A sync middleware in an async chain forces a thread switch.
• Long-lived connections + ORM. WebSocket consumer holding an open DB connection across awaits — connection state is fragile. Use database_sync_to_async (channels helper) to run DB calls in a thread with the right connection lifecycle.
• Transactions across await. Django transactions are tied to a thread; async with transaction.aatomic(): is new (5.x) and the right way. Don’t await inside a sync transaction.atomic() block.