Quick Takeaways
What you'll learn in this article
- 1
You're making concurrent external API calls (the asyncio.gather() pattern)
- 2
You need predictable latency under high concurrency
- 3
You're building LLM-powered features that benefit from concurrent inference calls
- 4
You're already on ASGI for Django Channels
- 5
Async Django handles concurrent HTTP requests efficiently
Keep reading for detailed implementation, code examples, and real-world results
Django's async story is one of the most ambitious infrastructure rewrites in web framework history. What started as DEP 0009 โ a formal proposal by Andrew Godwin in 2018 โ has evolved across seven major releases into a framework that can handle async views, async ORM queries, async signals, async authentication, and as of Django 6.0, even background task execution.
But here's the thing most articles won't tell you: only 14% of Django developers actually use async views in production. According to the 2024 Django Developer Survey by JetBrains and the Django Software Foundation, when Django developers need async capabilities, they're more likely to reach for FastAPI than Django's own async features.
That's not because Django's async implementation is bad. It's because most developers don't understand when it helps, when it hurts, and how to deploy it without shooting themselves in the foot.
This guide fixes that. We'll cover every async capability Django offers in 2026, show you real benchmarks that will surprise you, walk through production deployment patterns, and give you the practical knowledge to decide whether async Django belongs in your stack.
Django Async Adoption
14%
of Django developers use async views
If you're new to Django entirely, start with our Django 6 getting started tutorial first, then come back here for the async deep dive.
1. The Seven-Year Journey: How Django Went Async
Django was born synchronous. For 14 years โ from its first release in 2005 through 2019 โ every request followed the same pattern: arrive, execute sequentially, return. One thread, one request, start to finish. This model was simple, predictable, and powered some of the largest web applications on the planet including Instagram's 2+ billion user platform.
Then the world changed. WebSockets became essential for real-time features. External API calls became the bottleneck instead of database queries. AI inference endpoints needed to handle hundreds of concurrent connections without spawning hundreds of threads. The synchronous model that made Django reliable started making it inflexible.
Andrew Godwin โ the same developer who built Django Channels and the South migrations framework โ recognized this tension and published A Django Async Roadmap in June 2018. The philosophy was clear from the start:
"The plan is not to remove synchronous Django โ the plan is to keep that around, working as it does now, with asynchronous code being an option for those that feel they need the extra performance or flexibility."
The Django Technical Board formally approved DEP 0009 in July 2019, and the implementation began.
Django 3.0
ASGI protocol support โ the foundation layer. No async views yet.
Django 3.1
First async views and middleware. You could finally write 'async def my_view(request)'.
Django 4.0
Async cache interface โ aget(), aset(), adelete().
Django 4.1
Async ORM methods and async class-based views. The big one.
Django 4.2
Async streaming responses, async model methods (asave, adelete), psycopg 3 support.
Django 5.0
Async auth, async signals with concurrent dispatch, async prefetch.
Django 5.1
Async sessions, async-compatible decorators (login_required, etc.).
Django 5.2 LTS
Async auth backends, async user model methods and permissions.
Django 6.0
AsyncPaginator, built-in background tasks framework, Python 3.12+ required.
The Three-Stage Strategy
DEP 0009 specified a pragmatic approach for converting each Django component. Rather than rewriting everything at once, each subsystem would progress through three stages:
Stage 1 โ Sync-only: The existing state. Everything runs synchronously.
Stage 2 โ Sync-native with async wrapper: The synchronous code stays as-is, but gets wrapped with sync_to_async so it can be called from async contexts. This is where most of the ORM lives today.
Stage 3 โ Async-native with sync wrapper: The code is rewritten to be natively async, with async_to_sync wrappers for backward compatibility. This is the end goal, but only a few components have reached it.
This strategy meant Django could ship async features incrementally without breaking the millions of existing synchronous applications. It also meant that for many components โ particularly the ORM โ the async interface is a compatibility layer over synchronous code, not a ground-up async implementation. Understanding this distinction is critical for making informed performance decisions.
2. What's Async-Capable in Django 6.0 (and What Isn't)
As of Django 6.0 (released December 3, 2025), here's the honest status of every major subsystem:
Fully Async-Capable
Views have been async since Django 3.1. Both function-based and class-based views support async def. Every built-in decorator โ @login_required, @csrf_exempt, @cache_control, @require_http_methods โ works with async views as of Django 6.0.
Middleware supports async, but with a critical caveat we'll cover in section 5.
Signals gained async dispatch in Django 5.0 via Signal.asend() and Signal.asend_robust(). When you call asend(), all async receivers execute concurrently through asyncio.gather() โ a genuine performance win for signal-heavy applications.
Authentication is fully async: alogin(), alogout(), aauthenticate(), acheck_password(), and HttpRequest.auser(). Django 5.2 added async auth backends so the entire auth chain can stay async without context-switching.
Sessions โ all built-in session backends provide async APIs since Django 5.1.
Cache โ cache.aget(), cache.aset(), cache.adelete() since Django 4.0.
Pagination โ AsyncPaginator and AsyncPage are new in Django 6.0.
Background Tasks โ Django 6.0 introduced a built-in tasks framework with the @task decorator and task.enqueue(). Tasks run outside the request-response cycle in a separate worker process. This is Django's answer to simple Celery use cases.
Async-Wrapped (Not Natively Async)
The ORM is the most important component in this category. Every SQL-triggering QuerySet method has an a-prefixed async variant: aget(), acreate(), afilter() iteration, acount(), aexists(), abulk_create(), and more. But these methods are wrappers โ they use sync_to_async() internally to run the synchronous database code in a threadpool.
This means each async ORM call still consumes a thread. The concurrency benefit comes from the view-level event loop being able to handle other HTTP requests while waiting for that thread to finish. Andrew Godwin has acknowledged that fully native async ORM may be impossible, stating the team cannot "write and maintain what are two parallel ORM cores."
Model instance methods โ asave(), adelete(), arefresh_from_db() โ follow the same wrapper pattern.
Still Sync-Only
Database transactions have no async interface. You must wrap transactional code with sync_to_async:
from asgiref.sync import sync_to_async
from django.db import transaction
@sync_to_async(thread_sensitive=True)
@transaction.atomic
def transfer_funds(from_account, to_account, amount):
from_account.balance -= amount
from_account.save()
to_account.balance += amount
to_account.save()
return True
Template rendering โ no async template tags or filters. Templates execute synchronously regardless of whether the view is async.
Forms โ not addressed in the async roadmap. Forms remain entirely synchronous.
Management commands โ sync-only, though you can wrap async functions with async_to_sync():
from asgiref.sync import async_to_sync
from django.core.management.base import BaseCommand
class Command(BaseCommand):
def handle(self, *args, **options):
result = async_to_sync(self.do_async_work)()
self.stdout.write(f"Done: {result}")
async def do_async_work(self):
# Your async logic here
return "completed"
3. Writing Your First Async View
Let's move from theory to practice. Here's how async views work in Django 6.0.
Function-Based Async Views
The simplest case โ just add async before def:
# views.py
from django.http import JsonResponse
from .models import Article
async def article_list(request):
articles = []
async for article in Article.objects.filter(published=True).order_by('-created_at')[:20]:
articles.append({
'id': article.id,
'title': article.title,
'slug': article.slug,
})
return JsonResponse({'articles': articles})
Notice the async for โ this is how you iterate over QuerySets in async views. The QuerySet itself is still lazy (no SQL executes until iteration), but the iteration happens asynchronously.
For single-object retrieval, use await with the a-prefixed methods:
async def article_detail(request, slug):
try:
article = await Article.objects.select_related('author').aget(slug=slug)
except Article.DoesNotExist:
return JsonResponse({'error': 'Not found'}, status=404)
return JsonResponse({
'title': article.title,
'content': article.content,
'author': article.author.name,
})
The select_related('author') call is critical here. In async contexts, accessing a ForeignKey that hasn't been prefetched will raise a SynchronousOnlyOperation error. Always use select_related() or prefetch_related() explicitly in async views.
Class-Based Async Views
Django 4.1 added async support to class-based views. Your get(), post(), and other handler methods can be async:
from django.views import View
from django.http import JsonResponse
class DashboardView(View):
async def get(self, request):
import asyncio
# Run three database queries concurrently
user_count, active_sessions, recent_errors = await asyncio.gather(
User.objects.acount(),
Session.objects.filter(active=True).acount(),
ErrorLog.objects.filter(
created_at__gte=timezone.now() - timedelta(hours=1)
).acount(),
)
return JsonResponse({
'users': user_count,
'active_sessions': active_sessions,
'recent_errors': recent_errors,
})
This is where async Django starts to shine. Those three database queries execute concurrently via asyncio.gather() โ instead of waiting for each one sequentially, they all go to the threadpool simultaneously and the view returns when the slowest one finishes.
Note that __init__() and as_view() must remain synchronous. Only the HTTP handler methods can be async.
Concurrent External API Calls
The most compelling use case for async views isn't database queries โ it's calling external APIs. Consider a view that aggregates data from multiple services:
import asyncio
import httpx
async def aggregated_metrics(request):
async with httpx.AsyncClient(timeout=10.0) as client:
github_task = client.get(
'https://api.github.com/repos/myorg/myrepo/stats/contributors',
headers={'Authorization': f'token {settings.GITHUB_TOKEN}'}
)
sentry_task = client.get(
f'https://sentry.io/api/0/projects/{settings.SENTRY_ORG}/{settings.SENTRY_PROJECT}/stats/',
headers={'Authorization': f'Bearer {settings.SENTRY_TOKEN}'}
)
analytics_task = client.get(
f'https://analytics.example.com/api/v1/metrics',
headers={'X-API-Key': settings.ANALYTICS_KEY}
)
github_resp, sentry_resp, analytics_resp = await asyncio.gather(
github_task, sentry_task, analytics_task,
return_exceptions=True
)
# Process responses, handling any that failed
result = {}
if not isinstance(github_resp, Exception):
result['contributors'] = len(github_resp.json())
if not isinstance(sentry_resp, Exception):
result['error_rate'] = sentry_resp.json().get('error_rate')
if not isinstance(analytics_resp, Exception):
result['page_views'] = analytics_resp.json().get('total_views')
return JsonResponse(result)
In a synchronous view, these three API calls would take time_github + time_sentry + time_analytics. With async, they take max(time_github, time_sentry, time_analytics). If each call takes 200ms, that's 600ms sync vs 200ms async โ a 3x improvement.
This pattern is why companies like Kraken Technologies adopted async Django for their LLM-powered internal tools. When you're making multiple calls to language model APIs, the latency savings from concurrent execution are substantial.
4. The Async ORM: Every Method, Explained
Django 4.1 added async ORM methods via PR #14843. Here's the complete reference for Django 6.0.
Methods That Require await
These methods trigger SQL execution and must be awaited in async contexts:
| Category | Sync Method | Async Method |
|---|---|---|
| Single object | get() | aget() |
first() | afirst() | |
last() | alast() | |
earliest() | aearliest() | |
latest() | alatest() | |
| Create/Update | create() | acreate() |
get_or_create() | aget_or_create() | |
update_or_create() | aupdate_or_create() | |
bulk_create() | abulk_create() | |
bulk_update() | abulk_update() | |
update() | aupdate() | |
delete() | adelete() | |
| Aggregation | count() | acount() |
exists() | aexists() | |
contains() | acontains() | |
aggregate() | aaggregate() | |
in_bulk() | ain_bulk() | |
| Iteration | iterator() | aiterator() |
| Debug | explain() | aexplain() |
Methods That Don't Need Async Versions
These methods are lazy โ they build the query but don't execute SQL. They work identically in sync and async contexts:
filter(), exclude(), annotate(), order_by(), distinct(), select_related(), prefetch_related(), values(), values_list(), defer(), only(), union(), intersection(), difference()
You chain these normally, then use an async method to execute:
# The filter/order_by/values chain is lazy โ no SQL yet
queryset = (
Article.objects
.filter(status='published', category='engineering')
.order_by('-view_count')
.values('title', 'slug', 'view_count')[:10]
)
# SQL executes here, asynchronously
top_articles = await queryset.alist()
The alist() utility deserves special attention. It converts the async iterable to a list:
# These are equivalent: articles = [a async for a in Article.objects.filter(published=True)] articles = await Article.objects.filter(published=True).alist()
Model Instance Methods
Django 4.2 added async methods to model instances:
async def update_article(request, pk):
article = await Article.objects.aget(pk=pk)
article.title = "Updated Title"
article.view_count += 1
await article.asave()
return JsonResponse({'status': 'updated'})
async def remove_article(request, pk):
article = await Article.objects.aget(pk=pk)
await article.adelete()
return JsonResponse({'status': 'deleted'})
async def refresh_article(request, pk):
article = await Article.objects.aget(pk=pk)
# Re-fetch from database to get latest values
await article.arefresh_from_db()
return JsonResponse({'view_count': article.view_count})
The Concurrent Query Pattern
The most powerful pattern in async Django is running multiple ORM queries concurrently. Instead of sequential execution:
# Sequential โ each query waits for the previous one
# Total time: time_1 + time_2 + time_3
async def dashboard_slow(request):
users = await User.objects.acount()
orders = await Order.objects.filter(status='pending').acount()
revenue = await Order.objects.filter(
status='completed'
).aaggregate(total=Sum('amount'))
return JsonResponse({...})
Use asyncio.gather() or asyncio.TaskGroup() for concurrent execution:
# Concurrent โ all queries run at the same time
# Total time: max(time_1, time_2, time_3)
import asyncio
async def dashboard_fast(request):
users, orders, revenue = await asyncio.gather(
User.objects.acount(),
Order.objects.filter(status='pending').acount(),
Order.objects.filter(
status='completed'
).aaggregate(total=Sum('amount')),
)
return JsonResponse({
'users': users,
'orders': orders,
'revenue': revenue['total'],
})
For Python 3.12+ (required by Django 6.0), you can also use asyncio.TaskGroup() which provides better error handling:
async def dashboard_taskgroup(request):
async with asyncio.TaskGroup() as tg:
users_task = tg.create_task(User.objects.acount())
orders_task = tg.create_task(
Order.objects.filter(status='pending').acount()
)
revenue_task = tg.create_task(
Order.objects.filter(status='completed')
.aaggregate(total=Sum('amount'))
)
# All tasks guaranteed complete here
return JsonResponse({
'users': users_task.result(),
'orders': orders_task.result(),
'revenue': revenue_task.result()['total'],
})
If one task fails with TaskGroup, all other tasks are cancelled and the exception propagates. With gather(), you need return_exceptions=True to handle partial failures.
5. ASGI vs WSGI: The Deployment Decision
This is where most Django async guides get it wrong. They tell you to switch to ASGI without explaining the tradeoffs. Here's the reality.
WSGI vs ASGI Deployment
WSGI (Gunicorn)
ASGI (Uvicorn)
The Sync Middleware Trap
Here's the single most important deployment fact about async Django:
If any synchronous middleware sits between the ASGI server and your async view, Django switches into sync mode for the entire request.
This means one sync middleware can eliminate all the benefits of ASGI deployment. Django has to allocate a synchronous thread for the middleware, then context-switch back to async for the view, keeping that thread alive for exception propagation.
Check your middleware stack:
MIDDLEWARE = [
'django.middleware.security.SecurityMiddleware', # Async-capable
'django.contrib.sessions.middleware.SessionMiddleware', # Async-capable (5.1+)
'django.middleware.common.CommonMiddleware', # Async-capable
'django.middleware.csrf.CsrfViewMiddleware', # Async-capable
'django.contrib.auth.middleware.AuthenticationMiddleware', # Async-capable (5.0+)
'django.contrib.messages.middleware.MessageMiddleware', # Async-capable
'django.middleware.clickjacking.XFrameOptionsMiddleware', # Async-capable
]
All of Django's built-in middleware is async-capable as of Django 5.2. But third-party middleware might not be. Check every middleware in your stack.
For custom middleware, declare async capability explicitly:
class MyAsyncMiddleware:
async_capable = True
sync_capable = False # Set to True for dual-mode
def __init__(self, get_response):
self.get_response = get_response
async def __call__(self, request):
# Pre-processing
request.start_time = time.monotonic()
response = await self.get_response(request)
# Post-processing
duration = time.monotonic() - request.start_time
response['X-Request-Duration'] = f'{duration:.3f}s'
return response
ASGI Server Options
Three production-ready ASGI servers are available in 2026:
Uvicorn is the most popular choice. Built by the Encode team (who also created Starlette and httpx), it's lightweight, fast, and supports HTTP/1.1, HTTP/2, and WebSockets. Adding uvloop as the event loop backend provides a 2-4x throughput improvement over standard asyncio:
pip install uvicorn[standard] uvicorn myproject.asgi:application --host 0.0.0.0 --port 8000 --workers 4
Daphne is the official ASGI server maintained by the Django organization. It's built on Twisted and is the default choice for Django Channels projects. It supports HTTP/1.1 and WebSockets but not HTTP/2:
pip install daphne daphne -b 0.0.0.0 -p 8000 myproject.asgi:application
Hypercorn supports HTTP/1.1, HTTP/2, HTTP/3, and WebSockets, plus multiple async backends (asyncio, uvloop, trio). It's the right choice if you need HTTP/2+ support:
pip install hypercorn hypercorn myproject.asgi:application --bind 0.0.0.0:8000 --workers 4
Database Connection Configuration
This catches many teams during ASGI migration. Persistent database connections (CONN_MAX_AGE) don't work properly with ASGI because the event loop can process multiple requests on the same connection concurrently, causing race conditions.
Set CONN_MAX_AGE = 0 and use database-level connection pooling instead:
# settings.py
DATABASES = {
'default': {
'ENGINE': 'django.db.backends.postgresql',
'NAME': 'myapp',
'HOST': 'localhost',
'PORT': '5432',
'USER': 'myapp',
'PASSWORD': 'secret',
'CONN_MAX_AGE': 0, # Required for ASGI
'OPTIONS': {
'pool': True, # Django 5.1+ native connection pooling (psycopg 3)
},
}
}
The native connection pooling option (available since Django 5.1 with psycopg 3) can reduce database latency by 50-70ms according to benchmarks by Saurabh Kumar. This alone can make a bigger performance difference than async views for database-heavy applications.
6. Real-World Benchmarks: The Numbers That Matter
Let's talk about performance with actual data instead of hypothetical claims. The most rigorous Django async benchmarks available come from Hackeryarn, run on Python 3.14 with PostgreSQL 18, using the Granian server with 64 concurrent connections.
Static Content (No Database)
| framework | rps |
|---|---|
| Sync Django | 6614 |
| Async Django | 1120 |
| FastAPI | 37353 |
For raw throughput without database access, sync Django is 6x faster than async Django. FastAPI dominates at 37,353 RPS. This might seem like async Django is terrible, but these numbers measure the overhead of the async event loop machinery. In real applications, the bottleneck is almost never raw Python execution speed.
Database Reads (The Real World)
| config | rps |
|---|---|
| Sync Django | 669 |
| Sync + Pooling | 1822 |
| Async Django | 541 |
| FastAPI | 409 |
This is the benchmark that changes the conversation. When a database is involved โ which is every real Django application โ sync Django with connection pooling outperforms everything else by a wide margin. Sync Django with pooling achieves 1,822 RPS versus async Django's 541 RPS. FastAPI actually performs worst at 409 RPS.
Sync Django + Pooling
1,822 RPS
3.4x faster than async Django for DB reads
Database Writes
For write-heavy workloads (SELECT FOR UPDATE), all configurations converge to similar throughput (~160-170 RPS) because the database itself becomes the bottleneck. Async doesn't help when the limiting factor is disk I/O and row-level locks.
Where Async Actually Wins
The benchmarks reveal two scenarios where async Django provides clear benefits:
1. Tail Latency: Sync Django without pooling shows extreme tail latency โ the slowest requests take 100x longer than the average. Async Django maintains consistent latency with no massive outliers. For applications where predictable response times matter more than raw throughput, async provides stability.
2. Scaling with Workers: When adding more workers, async Django shows doubled throughput while sync Django sees diminishing returns. In highly distributed environments where the Django service is the bottleneck, async scaling characteristics become advantageous.
The Takeaway
For most Django applications, the highest-impact performance optimization is not switching to async. It's enabling connection pooling on sync Django. If you're on PostgreSQL with psycopg 3, add 'pool': True to your database config and you'll likely see a bigger improvement than any async migration.
Async Django makes sense when:
- You're making concurrent external API calls (the asyncio.gather() pattern)
- You need WebSocket support
- You need predictable latency under high concurrency
- You're building LLM-powered features that benefit from concurrent inference calls
- You're already on ASGI for Django Channels
For everything else, sync Django with connection pooling is faster and simpler. This aligns with what the Django core team intended โ async is an option for specific use cases, not a universal upgrade.
7. Async Authentication: The Full Chain
Django 5.0 through 5.2 built out the complete async authentication chain. Here's how it works end-to-end.
Async Login Flow
from django.contrib.auth import aauthenticate, alogin, alogout
async def login_view(request):
if request.method == 'POST':
username = request.POST.get('username')
password = request.POST.get('password')
# Async authentication โ checks credentials without blocking
user = await aauthenticate(request, username=username, password=password)
if user is not None:
await alogin(request, user)
return JsonResponse({'status': 'authenticated', 'user': user.username})
else:
return JsonResponse({'error': 'Invalid credentials'}, status=401)
async def logout_view(request):
await alogout(request)
return JsonResponse({'status': 'logged out'})
Async User Access in Views
Django 5.0 added HttpRequest.auser() through the AuthenticationMiddleware:
async def profile_view(request):
user = await request.auser()
if user.is_anonymous:
return JsonResponse({'error': 'Not authenticated'}, status=401)
return JsonResponse({
'username': user.username,
'email': user.email,
'date_joined': user.date_joined.isoformat(),
})
Async Auth Backends (Django 5.2+)
You can write fully async authentication backends that avoid sync-to-async context switching entirely:
from django.contrib.auth.backends import BaseBackend
class AsyncAPIBackend(BaseBackend):
"""Authenticate against an external API without blocking."""
async def aauthenticate(self, request, username=None, password=None, **kwargs):
import httpx
async with httpx.AsyncClient() as client:
response = await client.post(
'https://auth.example.com/verify',
json={'username': username, 'password': password}
)
if response.status_code == 200:
data = response.json()
user, _ = await User.objects.aget_or_create(
username=username,
defaults={'email': data.get('email', '')}
)
return user
return None
async def aget_user(self, user_id):
try:
return await User.objects.aget(pk=user_id)
except User.DoesNotExist:
return None
When Django's async auth functions detect that a backend implements aauthenticate() or aget_user(), they'll call the async versions directly instead of wrapping the sync versions.
Async-Compatible Decorators
All auth decorators work with async views in Django 6.0:
from django.contrib.auth.decorators import login_required, permission_required
@login_required
async def protected_view(request):
return JsonResponse({'message': 'You are authenticated'})
@permission_required('app.can_edit')
async def admin_view(request):
return JsonResponse({'message': 'You have edit permission'})
8. Async Signals: Concurrent Event Handling
Django 5.0 added Signal.asend() and Signal.asend_robust(), enabling async signal receivers that execute concurrently.
Why This Matters
In sync Django, signals fire sequentially. If you have three receivers connected to post_save, they execute one after another. If each takes 100ms (maybe they're sending emails, updating caches, or calling external APIs), the save operation takes an extra 300ms.
With async signals, all three receivers fire concurrently via asyncio.gather(), taking only 100ms total.
Writing Async Receivers
from django.db.models.signals import post_save
from django.dispatch import receiver
import httpx
@receiver(post_save, sender=Article)
async def notify_subscribers(sender, instance, created, **kwargs):
if created:
subscribers = [
sub async for sub in
Subscription.objects.filter(
category=instance.category, active=True
).select_related('user')
]
async with httpx.AsyncClient() as client:
tasks = [
client.post(
'https://api.email-service.com/send',
json={
'to': sub.user.email,
'subject': f'New article: {instance.title}',
'body': instance.excerpt,
}
)
for sub in subscribers
]
await asyncio.gather(*tasks, return_exceptions=True)
@receiver(post_save, sender=Article)
async def update_search_index(sender, instance, **kwargs):
async with httpx.AsyncClient() as client:
await client.put(
f'https://search.example.com/articles/{instance.id}',
json={'title': instance.title, 'content': instance.content}
)
@receiver(post_save, sender=Article)
async def invalidate_cache(sender, instance, **kwargs):
from django.core.cache import cache
await cache.adelete(f'article:{instance.slug}')
await cache.adelete('article:list')
Dispatching Async Signals
# From an async context, use asend()
results = await my_signal.asend(sender=MyClass, data=my_data)
# asend_robust() catches exceptions from individual receivers
results = await my_signal.asend_robust(sender=MyClass, data=my_data)
for receiver, response in results:
if isinstance(response, Exception):
logger.error(f"Signal receiver {receiver} failed: {response}")
When asend() is called, async receivers run concurrently and sync receivers are called sequentially in a threadpool. Mixed signal handlers work correctly โ you don't need to convert all receivers at once.
9. Django Channels and WebSockets
Django Channels extends Django beyond HTTP to handle WebSockets, background workers, and other long-lived connections. It predates Django's native async support and remains the standard for real-time Django applications.
The Relationship Between Channels and Async Django
Django Channels and async Django are complementary:
- Async Django handles concurrent HTTP requests efficiently
- Django Channels handles non-HTTP protocols (WebSockets, MQTT, etc.)
- Both run on ASGI
- Channels consumers can use async ORM methods directly
Building an Async WebSocket Consumer
# consumers.py
import json
from channels.generic.websocket import AsyncWebSocketConsumer
class LiveDashboardConsumer(AsyncWebSocketConsumer):
async def connect(self):
self.room_name = self.scope['url_route']['kwargs']['dashboard_id']
self.room_group = f'dashboard_{self.room_name}'
# Join the channel group
await self.channel_layer.group_add(
self.room_group,
self.channel_name
)
await self.accept()
# Send current stats on connection
stats = await self.get_dashboard_stats()
await self.send(text_data=json.dumps({
'type': 'initial_stats',
'data': stats
}))
async def disconnect(self, close_code):
await self.channel_layer.group_discard(
self.room_group,
self.channel_name
)
async def receive(self, text_data=None):
data = json.loads(text_data)
if data.get('action') == 'refresh':
stats = await self.get_dashboard_stats()
await self.send(text_data=json.dumps({
'type': 'stats_update',
'data': stats
}))
async def stats_update(self, event):
"""Handle messages from the channel group."""
await self.send(text_data=json.dumps(event['data']))
async def get_dashboard_stats(self):
import asyncio
users, orders, errors = await asyncio.gather(
User.objects.acount(),
Order.objects.filter(status='pending').acount(),
ErrorLog.objects.filter(
created_at__gte=timezone.now() - timedelta(minutes=5)
).acount(),
)
return {
'active_users': users,
'pending_orders': orders,
'recent_errors': errors,
}
Routing Configuration
# routing.py
from django.urls import re_path
from . import consumers
websocket_urlpatterns = [
re_path(
r'ws/dashboard/(?P<dashboard_id>\w+)/$',
consumers.LiveDashboardConsumer.as_asgi()
),
]
# asgi.py
import os
from django.core.asgi import get_asgi_application
from channels.routing import ProtocolTypeRouter, URLRouter
from channels.auth import AuthMiddlewareStack
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myproject.settings')
application = ProtocolTypeRouter({
'http': get_asgi_application(),
'websocket': AuthMiddlewareStack(
URLRouter(
# Import your routing here
websocket_urlpatterns
)
),
})
Database Connections in Long-Lived Consumers
WebSocket consumers can live for hours or days. Database connections opened during the consumer's lifetime can go stale. Always close old connections before making queries:
from django.db import aclose_old_connections
class LongLivedConsumer(AsyncWebSocketConsumer):
async def receive(self, text_data=None):
# Clean up stale database connections before querying
await aclose_old_connections()
# Now safe to query
data = await MyModel.objects.aget(pk=some_id)
await self.send(text_data=json.dumps({'data': str(data)}))
This is especially important if you're using CONN_MAX_AGE = 0 (which you should be on ASGI).
10. Async Streaming Responses and Server-Sent Events
Django 4.2 added async iterator support to StreamingHttpResponse, opening the door to Server-Sent Events (SSE) โ a lightweight alternative to WebSockets for one-way real-time updates.
Why SSE Over WebSockets
WebSockets provide bidirectional communication, but many real-time use cases only need server-to-client updates: live dashboards, progress indicators, log streaming, AI response streaming. For these, SSE is simpler:
- Works over standard HTTP โ no protocol upgrade needed
- Automatic reconnection built into the browser's EventSource API
- Works through HTTP proxies and load balancers without special configuration
- No need for Django Channels or a channel layer
Building an SSE Endpoint
import asyncio
import json
from django.http import StreamingHttpResponse
async def sse_dashboard_updates(request):
async def event_stream():
while True:
# Fetch latest metrics
stats = await get_live_stats()
# Format as SSE
data = json.dumps(stats)
yield f"data: {data}\n\n"
# Wait before next update
await asyncio.sleep(2)
response = StreamingHttpResponse(
event_stream(),
content_type='text/event-stream'
)
response['Cache-Control'] = 'no-cache'
response['X-Accel-Buffering'] = 'no' # Disable nginx buffering
return response
async def get_live_stats():
import asyncio
active_users, cpu_usage, error_count = await asyncio.gather(
Session.objects.filter(
last_activity__gte=timezone.now() - timedelta(minutes=5)
).acount(),
get_system_metrics(),
ErrorLog.objects.filter(
timestamp__gte=timezone.now() - timedelta(minutes=1)
).acount(),
)
return {
'active_users': active_users,
'cpu_usage': cpu_usage,
'errors_per_minute': error_count,
'timestamp': timezone.now().isoformat(),
}
The client-side code is minimal:
const eventSource = new EventSource('/api/dashboard/stream/')
eventSource.onmessage = event => {
const data = JSON.parse(event.data)
document.getElementById('active-users').textContent = data.active_users
document.getElementById('cpu-usage').textContent = `${data.cpu_usage}%`
document.getElementById('error-count').textContent = data.errors_per_minute
}
eventSource.onerror = () => {
console.log('Connection lost. Reconnecting...')
// EventSource automatically reconnects
}
Streaming AI Responses
One of the most compelling SSE use cases in 2026 is streaming LLM responses. Instead of waiting for the entire response to generate, you can stream tokens as they arrive:
import httpx
async def stream_ai_response(request):
prompt = request.GET.get('prompt', '')
async def token_stream():
async with httpx.AsyncClient() as client:
async with client.stream(
'POST',
'https://api.anthropic.com/v1/messages',
headers={
'x-api-key': settings.ANTHROPIC_API_KEY,
'anthropic-version': '2023-06-01',
'content-type': 'application/json',
},
json={
'model': 'claude-sonnet-4-5-20250929',
'max_tokens': 1024,
'stream': True,
'messages': [{'role': 'user', 'content': prompt}],
},
) as response:
async for line in response.aiter_lines():
if line.startswith('data: '):
data = json.loads(line[6:])
if data.get('type') == 'content_block_delta':
text = data['delta'].get('text', '')
yield f"data: {json.dumps({'text': text})}\n\n"
yield "data: {\"done\": true}\n\n"
return StreamingHttpResponse(
token_stream(),
content_type='text/event-stream'
)
This pattern is exactly why Kraken Technologies adopted async Django for their LLM-powered internal tools. The async streaming response keeps the connection open and forwards tokens to the client as they arrive from the AI provider, without blocking a thread for the entire inference duration.
SSE vs WebSockets: Decision Matrix
Real-Time Communication Patterns
Server-Sent Events (SSE)
WebSockets (Channels)
11. Django 6.0's Background Tasks Framework
Django 6.0 introduced a built-in task execution framework โ Django's answer to simple Celery use cases. Instead of setting up a separate task queue infrastructure, you can define background tasks directly in your Django application.
Defining Tasks
# tasks.py
from django.tasks import task
@task()
def send_welcome_email(user_id):
user = User.objects.get(pk=user_id)
send_email(
to=user.email,
subject='Welcome!',
body=f'Welcome to our platform, {user.first_name}!'
)
@task(priority=10)
def generate_report(report_type, date_range):
# Long-running report generation
data = aggregate_data(report_type, date_range)
report = format_report(data)
Report.objects.create(
type=report_type,
content=report,
status='completed'
)
Enqueuing Tasks from Views
from .tasks import send_welcome_email, generate_report
async def register_user(request):
user = await User.objects.acreate(
username=request.POST['username'],
email=request.POST['email'],
)
# Enqueue background task โ returns immediately
send_welcome_email.enqueue(user.id)
return JsonResponse({'status': 'registered', 'user_id': user.id})
async def request_report(request):
generate_report.enqueue(
report_type=request.GET['type'],
date_range=request.GET['range']
)
return JsonResponse({'status': 'report generation started'})
Tasks are executed by a separate worker process. For development, Django includes a simple worker, but for production you'll want a more robust backend. The framework is designed to be backend-agnostic โ similar to how Django's cache framework supports multiple backends.
This doesn't replace Celery for complex workflows (periodic tasks, task chains, rate limiting, retries with backoff), but for simple "fire and forget" tasks it eliminates a significant infrastructure dependency.
11. Production Patterns and Pitfalls
Let's cover the real-world lessons that teams like Kraken Technologies have documented after running async Django in production.
The Lazy QuerySet Trap
This is the most common bug in async Django code:
# WRONG โ QuerySet evaluation is deferred
all_users = await sync_to_async(User.objects.all)()
# Later iteration fails because it happens in the wrong context
for user in all_users: # SynchronousOnlyOperation!
print(user.name)
# CORRECT โ evaluate the QuerySet immediately
all_users = await User.objects.all().alist()
for user in all_users:
print(user.name)
The Foreign Key Access Trap
In async contexts, accessing a ForeignKey that hasn't been prefetched raises an error:
# WRONG โ lazy FK access doesn't work in async
article = await Article.objects.aget(pk=1)
author_name = article.author.name # SynchronousOnlyOperation!
# CORRECT โ use select_related
article = await Article.objects.select_related('author').aget(pk=1)
author_name = article.author.name # Works!
# CORRECT โ for many-to-many or reverse relations
articles = await Article.objects.prefetch_related('tags').alist()
for article in articles:
tags = list(article.tags.all()) # Pre-fetched, works!
Transaction Patterns
Since @transaction.atomic is sync-only, here's the production pattern for transactional operations in async views:
from asgiref.sync import sync_to_async
from django.db import transaction
@sync_to_async(thread_sensitive=True)
@transaction.atomic
def create_order_with_items(order_data, items_data):
order = Order.objects.create(**order_data)
for item in items_data:
OrderItem.objects.create(order=order, **item)
return order
async def place_order(request):
order_data = json.loads(request.body)
order = await create_order_with_items(
order_data={'user_id': request.user.id, 'total': order_data['total']},
items_data=order_data['items']
)
return JsonResponse({'order_id': order.id})
The thread_sensitive=True parameter ensures all database operations within the transaction run on the same thread โ critical for transaction integrity.
Testing Async Views
# tests.py
import pytest
from django.test import AsyncClient
@pytest.mark.django_db(transaction=True)
@pytest.mark.asyncio
async def test_article_list():
# Create test data
await Article.objects.acreate(title='Test', slug='test', published=True)
client = AsyncClient()
response = await client.get('/api/articles/')
assert response.status_code == 200
data = response.json()
assert len(data['articles']) == 1
assert data['articles'][0]['title'] == 'Test'
@pytest.mark.django_db
@pytest.mark.asyncio
async def test_concurrent_queries():
import asyncio
# Verify gather works with ORM
count1, count2 = await asyncio.gather(
Article.objects.acount(),
User.objects.acount(),
)
assert isinstance(count1, int)
assert isinstance(count2, int)
Use pytest-asyncio with asyncio_mode = "auto" and pytest-django. The AsyncClient is Django's built-in async test client that sends requests through the ASGI stack.
The DJANGO_ALLOW_ASYNC_UNSAFE Footgun
You might encounter this environment variable in development when using Jupyter notebooks or the Django shell:
os.environ['DJANGO_ALLOW_ASYNC_UNSAFE'] = 'true'
Never set this in production. It disables Django's safety checks that prevent synchronous ORM access from async contexts. In production, this can cause data corruption, race conditions, and connection pool exhaustion. It exists solely for interactive development environments where thread safety is not a concern.
Client Disconnect Handling
Long-running async views should handle client disconnects gracefully:
import asyncio
async def long_running_view(request):
try:
result = await asyncio.wait_for(
expensive_computation(),
timeout=30.0
)
return JsonResponse({'result': result})
except asyncio.TimeoutError:
return JsonResponse({'error': 'Request timed out'}, status=504)
except asyncio.CancelledError:
# Client disconnected โ clean up resources
logger.info('Client disconnected during long-running request')
raise
12. Case Study: Kraken Technologies' Async Django in Production
Kraken Technologies โ the technology arm of Octopus Energy โ published the most detailed public case study of async Django in production in January 2026. Their experience is instructive because it's honest about both the benefits and the pain points.
The Use Case
Kraken's team built LLM-powered internal tools: Slack chatbots that answer questions about internal documentation, automated release note generation, and code search tools. These tools make multiple concurrent calls to language model APIs, making async Django a natural fit.
They run Django 5.2 (the LTS release) deployed with Uvicorn, using PostgreSQL with CONN_MAX_AGE = 0.
What Worked Well
Concurrent API calls provided the biggest win. Their chatbot views make simultaneous calls to different LLM providers and internal APIs using asyncio.TaskGroup(). A view that previously took 3-4 seconds (sequential API calls) now completes in 800ms-1.2 seconds.
Async signal handlers allowed them to fire off analytics events and audit logging without blocking the response. When a user asks the chatbot a question, the response returns immediately while async signal receivers handle logging, usage tracking, and feedback collection concurrently.
The AsyncClient for testing let them write natural async test code that exercised the full ASGI stack, catching integration issues that unit tests missed.
What Caused Pain
Foreign key access was the most frequent source of SynchronousOnlyOperation errors during development. Developers accustomed to Django's lazy FK loading kept forgetting to add select_related(). They solved this with a custom linting rule that flags FK access in async views without explicit prefetching.
Transaction management required wrapping every transactional operation in @sync_to_async(thread_sensitive=True) combined with @transaction.atomic. The verbosity of this pattern was a recurring complaint:
# What they wanted to write:
async def update_conversation(conversation_id, message):
async with transaction.aatomic(): # Does not exist yet
conversation = await Conversation.objects.aget(pk=conversation_id)
await Message.objects.acreate(conversation=conversation, text=message)
conversation.last_activity = timezone.now()
await conversation.asave()
# What they actually had to write:
@sync_to_async(thread_sensitive=True)
@transaction.atomic
def _update_conversation_sync(conversation_id, message):
conversation = Conversation.objects.get(pk=conversation_id)
Message.objects.create(conversation=conversation, text=message)
conversation.last_activity = timezone.now()
conversation.save()
async def update_conversation(conversation_id, message):
await _update_conversation_sync(conversation_id, message)
Test factories didn't work out of the box. They had to create custom AsyncDjangoModelFactory classes to use with factory_boy in async test contexts.
Management commands remained synchronous. They wrapped async business logic with async_to_sync(), which worked but felt awkward.
Their Verdict
Kraken's overall assessment was "okay" โ async Django works and provides real benefits for their specific use case (concurrent LLM calls), but the developer experience is notably rougher than synchronous Django. They recommended async for teams that have specific, measurable I/O bottlenecks and are willing to invest in developer education about async patterns.
They did not recommend async for teams building standard CRUD applications or those who prefer the simplicity and predictability of synchronous code. This aligns with the broader community sentiment โ async Django is a targeted tool, not a universal upgrade.
13. Django Async vs FastAPI: When to Choose What
This comparison comes up in every Django async discussion. Here's an honest assessment based on benchmarks and real-world usage.
| Name | Value |
|---|---|
| Django (sync) | 52 |
| FastAPI | 28 |
| Django (async) | 14 |
| Flask | 6 |
Python web framework usage among professional developers (JetBrains/Django Survey 2024)
Django's Advantages
Ecosystem maturity: Django's ORM, admin interface, auth system, migrations framework, and 20+ years of third-party packages represent thousands of hours of solved problems. Building the equivalent infrastructure with FastAPI means assembling SQLAlchemy + Alembic + custom admin + custom auth + custom permissions โ each requiring its own learning curve.
Gradual migration: Existing Django applications can add async views incrementally. You don't rewrite your entire application โ you identify the specific views that benefit from async (external API calls, concurrent queries) and convert just those. Everything else stays synchronous.
Battle-tested at scale: Instagram runs the world's largest Django deployment serving 2+ billion users. Pinterest, Mozilla, NASA, Disqus, and Bitbucket all run Django in production. This level of production validation doesn't exist for any other Python async framework.
FastAPI's Advantages
Performance for pure APIs: For lightweight JSON API endpoints without Django's middleware overhead, FastAPI is significantly faster. If you're building a microservice that doesn't need Django's features, FastAPI's lean architecture makes sense.
Native async from the ground up: FastAPI was designed async-first. There's no wrapper layer, no sync-to-async overhead, no risk of accidentally using a sync middleware. The async experience is seamless.
Automatic OpenAPI documentation: FastAPI generates interactive API documentation from your type annotations. Django requires django-rest-framework plus drf-spectacular or similar for equivalent functionality.
The Hybrid Pattern
Many production systems in 2026 use both. As noted in framework comparison analyses, teams use Django for the "control plane" โ dashboards, user management, admin interfaces, content management โ and FastAPI for the "data plane" โ high-throughput API endpoints, AI inference services, real-time data processing.
This hybrid approach lets each framework play to its strengths. If you're interested in the Python production ecosystem beyond Django, our tutorial on building production-ready systems with Python and FastAPI covers the FastAPI side in depth.
14. Native Async Database Access with psycopg 3
For applications that need maximum database performance, Django 4.2+ supports psycopg 3 โ the modern PostgreSQL adapter with native async support.
While Django's ORM async methods use sync_to_async wrappers, you can bypass the ORM entirely for performance-critical queries and use psycopg 3's native async cursors:
import psycopg
async def raw_async_query(request):
conninfo = (
f"host={settings.DATABASES['default']['HOST']} "
f"port={settings.DATABASES['default']['PORT']} "
f"dbname={settings.DATABASES['default']['NAME']} "
f"user={settings.DATABASES['default']['USER']} "
f"password={settings.DATABASES['default']['PASSWORD']}"
)
async with await psycopg.AsyncConnection.connect(conninfo) as conn:
async with conn.cursor() as cur:
await cur.execute("""
SELECT a.title, a.view_count, u.username
FROM articles_article a
JOIN auth_user u ON a.author_id = u.id
WHERE a.status = 'published'
ORDER BY a.view_count DESC
LIMIT 10
""")
rows = await cur.fetchall()
return JsonResponse({
'top_articles': [
{'title': row[0], 'views': row[1], 'author': row[2]}
for row in rows
]
})
This bypasses all ORM overhead and the sync_to_async threadpool, giving you true async database I/O. The tradeoff is losing Django's ORM abstractions โ you're writing raw SQL.
For most applications, the ORM's async wrappers are sufficient. Reserve raw psycopg 3 for hot paths where you've profiled and confirmed the ORM overhead matters.
15. Monitoring and Observability for Async Django
Async Django introduces new observability challenges. Traditional Django monitoring tools assume synchronous, one-thread-per-request execution. With async views, a single thread handles multiple requests concurrently, and the relationship between threads, coroutines, and database connections becomes less intuitive.
Tracking Concurrent Request Volume
Unlike WSGI where worker count equals maximum concurrent requests, ASGI workers can handle hundreds of concurrent connections each. You need to track both worker utilization and event loop saturation:
import asyncio
import time
from django.http import JsonResponse
class AsyncMetricsMiddleware:
"""Track async-specific performance metrics."""
async_capable = True
sync_capable = False
def __init__(self, get_response):
self.get_response = get_response
self.active_requests = 0
self.total_requests = 0
self.total_latency = 0.0
async def __call__(self, request):
self.active_requests += 1
self.total_requests += 1
start = time.monotonic()
try:
response = await self.get_response(request)
finally:
duration = time.monotonic() - start
self.total_latency += duration
self.active_requests -= 1
# Emit metrics (Prometheus, Datadog, etc.)
response['X-Request-Duration'] = f'{duration:.4f}'
response['X-Active-Requests'] = str(self.active_requests)
return response
Event Loop Monitoring
A saturated event loop is the async equivalent of thread pool exhaustion. Monitor it by measuring how long scheduled callbacks take to execute:
import asyncio
import logging
logger = logging.getLogger('async_monitor')
async def monitor_event_loop(interval=5.0):
"""
Periodically checks event loop responsiveness.
If the loop is saturated, the callback will be delayed.
"""
while True:
loop = asyncio.get_event_loop()
scheduled_time = loop.time()
await asyncio.sleep(interval)
actual_delay = loop.time() - scheduled_time - interval
if actual_delay > 0.1: # More than 100ms behind schedule
logger.warning(
f"Event loop lag: {actual_delay:.3f}s "
f"(scheduled {interval}s sleep took "
f"{interval + actual_delay:.3f}s)"
)
Database Connection Pool Monitoring
With CONN_MAX_AGE = 0 and psycopg 3's connection pooling, you need to monitor pool utilization rather than per-thread connection counts:
from django.db import connection
async def connection_pool_stats(request):
"""Expose connection pool metrics for monitoring."""
pool = connection.pool # psycopg 3 pool reference
return JsonResponse({
'pool_size': pool.get_size(),
'pool_available': pool.get_available(),
'pool_usage_percent': (
(pool.get_size() - pool.get_available())
/ max(pool.get_size(), 1) * 100
),
'requests_waiting': pool.get_waiting(),
})
Structured Logging for Async Contexts
Standard Django logging loses context when multiple requests share a thread. Use contextvars to maintain request-specific context across async operations:
import contextvars
import uuid
import logging
request_id_var = contextvars.ContextVar('request_id', default=None)
class AsyncRequestIDMiddleware:
"""Assigns a unique ID to each request for log correlation."""
async_capable = True
def __init__(self, get_response):
self.get_response = get_response
async def __call__(self, request):
request_id = str(uuid.uuid4())[:8]
request_id_var.set(request_id)
request.request_id = request_id
response = await self.get_response(request)
response['X-Request-ID'] = request_id
return response
class RequestIDFilter(logging.Filter):
"""Injects request_id into all log records."""
def filter(self, record):
record.request_id = request_id_var.get('unknown')
return True
Configure in settings:
LOGGING = {
'version': 1,
'filters': {
'request_id': {
'()': 'myapp.logging.RequestIDFilter',
},
},
'formatters': {
'verbose': {
'format': '[{asctime}] [{request_id}] {levelname} {name}: {message}',
'style': '{',
},
},
'handlers': {
'console': {
'class': 'logging.StreamHandler',
'formatter': 'verbose',
'filters': ['request_id'],
},
},
'root': {
'handlers': ['console'],
'level': 'INFO',
},
}
With this setup, every log line from an async view includes the request ID, making it possible to trace a single request across concurrent coroutines, database queries, and external API calls. This is essential for debugging production issues where multiple requests are being processed simultaneously on the same thread.
Key Metrics to Track
| metric | importance |
|---|---|
| Event Loop Lag | 95 |
| Active Connections | 90 |
| DB Pool Usage | 88 |
| p99 Latency | 85 |
| Sync-to-Async Overhead | 75 |
| Thread Pool Size | 70 |
Event loop lag is the most critical async-specific metric. If it exceeds 100ms consistently, your async views are doing too much CPU-bound work on the event loop. Move heavy computation to sync_to_async with thread_sensitive=False or to the background tasks framework.
Database connection pool utilization tells you whether you need more pool capacity. If you consistently see requests waiting for connections, increase the pool size or optimize query patterns.
p99 latency is more important than average latency for async applications. Async Django's strength is consistent latency under load โ if your p99 is spiking, something is blocking the event loop.
16. Python 3.14 Free-Threading: The Next Chapter
Python 3.14's free-threading mode (the no-GIL implementation, PEP 779) is no longer experimental. This fundamentally changes the async calculus for Django.
With free-threading enabled (python --disable-gil), multiple threads can execute Python bytecode truly in parallel. The performance overhead has been reduced from the initial ~40% slowdown to approximately 5-10% on single-threaded workloads.
What this means for Django:
Sync views get faster: The traditional one-thread-per-request model can now fully utilize multiple CPU cores without the GIL bottleneck. CPU-bound operations in sync views see real parallelism.
The async wrapper overhead decreases: Django's sync_to_async ORM wrappers currently incur thread scheduling overhead. With free-threading, thread creation and context-switching become cheaper.
Thread safety becomes more critical: Code that was accidentally thread-safe because the GIL prevented true concurrent execution will now need explicit synchronization. Django's core is thread-safe, but third-party packages may not be.
A DjangoCon US 2025 talk on Free-Threaded Django explored these implications. Early benchmarks show promising results for Django view-level parallelism, but production adoption in 2026 remains cautious โ most teams are testing in staging environments before committing.
The convergence of async Django and free-threaded Python will likely reshape best practices over the next two years. For now, async Django gives you concurrency at the event loop level, and free-threading will eventually add CPU parallelism on top. They're complementary, not competing approaches.
16. Async Error Handling and Resilience Patterns
Production async Django applications need robust error handling. Here are patterns that go beyond basic try/catch.
Structured Error Handling in Async Views
import asyncio
import logging
from django.http import JsonResponse
from functools import wraps
logger = logging.getLogger(__name__)
def async_error_handler(view_func):
"""Decorator that wraps async views with structured error handling."""
@wraps(view_func)
async def wrapper(request, *args, **kwargs):
try:
return await view_func(request, *args, **kwargs)
except asyncio.CancelledError:
# Client disconnected โ log but don't treat as error
logger.info(
f"Client disconnected: {request.method} {request.path}"
)
raise
except asyncio.TimeoutError:
logger.warning(
f"Timeout: {request.method} {request.path}"
)
return JsonResponse(
{'error': 'Request timed out'},
status=504
)
except Exception:
logger.exception(
f"Unhandled error: {request.method} {request.path}"
)
return JsonResponse(
{'error': 'Internal server error'},
status=500
)
return wrapper
Timeout-Protected Concurrent Operations
When running multiple async operations concurrently, individual failures shouldn't take down the entire request:
async def resilient_dashboard(request):
"""Dashboard that degrades gracefully when services are slow."""
async def fetch_with_timeout(coro, timeout_seconds, default):
try:
return await asyncio.wait_for(coro, timeout=timeout_seconds)
except asyncio.TimeoutError:
logger.warning(f"Timeout fetching {coro.__name__}")
return default
except Exception as e:
logger.error(f"Error fetching {coro.__name__}: {e}")
return default
users, orders, analytics = await asyncio.gather(
fetch_with_timeout(User.objects.acount(), 2.0, default=None),
fetch_with_timeout(
Order.objects.filter(status='pending').acount(),
2.0,
default=None
),
fetch_with_timeout(
fetch_external_analytics(),
5.0,
default={'page_views': None, 'bounce_rate': None}
),
)
return JsonResponse({
'users': users,
'pending_orders': orders,
'analytics': analytics,
'degraded': any(v is None for v in [users, orders]),
})
Retry Logic for External Services
import asyncio
import httpx
async def fetch_with_retry(url, max_retries=3, backoff_base=0.5):
"""Fetch URL with exponential backoff retry."""
last_exception = None
for attempt in range(max_retries):
try:
async with httpx.AsyncClient(timeout=10.0) as client:
response = await client.get(url)
response.raise_for_status()
return response.json()
except (httpx.HTTPStatusError, httpx.ConnectError) as e:
last_exception = e
if attempt < max_retries - 1:
wait_time = backoff_base * (2 ** attempt)
logger.warning(
f"Retry {attempt + 1}/{max_retries} for {url} "
f"after {wait_time}s: {e}"
)
await asyncio.sleep(wait_time)
raise last_exception
Circuit Breaker Pattern
For services that may be completely down, a circuit breaker prevents cascading failures:
import asyncio
import time
class AsyncCircuitBreaker:
def __init__(self, failure_threshold=5, reset_timeout=30):
self.failure_threshold = failure_threshold
self.reset_timeout = reset_timeout
self.failure_count = 0
self.last_failure_time = 0
self.state = 'closed' # closed, open, half-open
async def call(self, coro, fallback=None):
if self.state == 'open':
if time.monotonic() - self.last_failure_time > self.reset_timeout:
self.state = 'half-open'
else:
if fallback is not None:
return fallback
raise RuntimeError("Circuit breaker is open")
try:
result = await coro
if self.state == 'half-open':
self.state = 'closed'
self.failure_count = 0
return result
except Exception as e:
self.failure_count += 1
self.last_failure_time = time.monotonic()
if self.failure_count >= self.failure_threshold:
self.state = 'open'
logger.error(
f"Circuit breaker opened after "
f"{self.failure_count} failures"
)
if fallback is not None:
return fallback
raise
# Usage
analytics_breaker = AsyncCircuitBreaker(failure_threshold=3, reset_timeout=60)
async def get_analytics(request):
data = await analytics_breaker.call(
fetch_external_analytics(),
fallback={'status': 'unavailable', 'cached': True}
)
return JsonResponse(data)
These patterns become essential when your async views depend on external services. The combination of timeouts, retries, and circuit breakers creates resilient applications that degrade gracefully instead of cascading failures across your infrastructure.
17. Migration Checklist: Going Async in Production
If you've read this far and decided async Django is right for your use case, here's a practical migration checklist:
Phase 1: Foundation
- Upgrade to Django 6.0 and Python 3.12+
- Switch to psycopg 3 with native connection pooling ('pool': True)
- Set CONN_MAX_AGE = 0 in database configuration
- Audit all middleware for async_capable = True
- Remove or replace any sync-only third-party middleware
Phase 2: ASGI Deployment
- Create or update asgi.py in your project
- Deploy behind Uvicorn (or Daphne/Hypercorn) alongside your existing WSGI deployment
- Run both in parallel to compare behavior
- Monitor thread pool utilization and connection counts
Phase 3: Incremental View Conversion
- Identify views that make external API calls โ convert these first (highest ROI)
- Convert views that run multiple independent database queries (use asyncio.gather())
- Convert views that need to check presence channels or send real-time updates
- Leave simple CRUD views synchronous โ they won't benefit from async
Phase 4: Advanced Features
- Add async signal receivers for I/O-bound event handlers
- Implement async auth backends if authenticating against external services
- Consider Django Channels for WebSocket requirements
- Evaluate the built-in tasks framework for simple background work
Phase 5: Monitoring
- Track per-view response times to measure async benefits
- Monitor database connection pool utilization
- Watch for SynchronousOnlyOperation errors in logs
- Profile sync_to_async overhead in hot paths
The key principle: convert incrementally, measure constantly. Don't rewrite your entire application. Identify the 20% of views where async provides 80% of the benefit, convert those, and leave the rest synchronous.
For teams working on database-heavy Django applications, our guide to advanced database performance optimization patterns covers the SQL-level optimizations that complement async view-level improvements.
18. Async Patterns Cookbook
Here are production-ready patterns for common async Django scenarios that you can copy and adapt.
Pattern 1: Async API Gateway View
When your Django application acts as an API gateway aggregating data from multiple microservices:
import asyncio
import httpx
from django.http import JsonResponse
SERVICES = {
'users': 'https://users-service.internal/api/v1',
'orders': 'https://orders-service.internal/api/v1',
'inventory': 'https://inventory-service.internal/api/v1',
'analytics': 'https://analytics-service.internal/api/v1',
}
async def aggregate_customer_view(request, customer_id):
"""
Fetch customer data from 4 microservices concurrently.
Sequential: ~800ms (4 x 200ms average)
Concurrent: ~200ms (max of the 4 calls)
"""
async with httpx.AsyncClient(timeout=5.0) as client:
tasks = {
'profile': client.get(
f"{SERVICES['users']}/customers/{customer_id}"
),
'orders': client.get(
f"{SERVICES['orders']}/customers/{customer_id}/recent"
),
'inventory': client.get(
f"{SERVICES['inventory']}/customers/{customer_id}/wishlist"
),
'metrics': client.get(
f"{SERVICES['analytics']}/customers/{customer_id}/engagement"
),
}
results = {}
responses = await asyncio.gather(
*tasks.values(),
return_exceptions=True
)
for key, response in zip(tasks.keys(), responses):
if isinstance(response, Exception):
results[key] = {'error': str(response)}
elif response.status_code == 200:
results[key] = response.json()
else:
results[key] = {'error': f'HTTP {response.status_code}'}
return JsonResponse(results)
Pattern 2: Async Bulk Import with Progress Tracking
For long-running import operations that report progress via the database:
from django.tasks import task
@task()
def import_csv_data(import_job_id, file_path):
"""Background task for CSV import with progress updates."""
import csv
job = ImportJob.objects.get(pk=import_job_id)
job.status = 'processing'
job.save()
with open(file_path, 'r') as f:
reader = csv.DictReader(f)
rows = list(reader)
total = len(rows)
batch = []
for i, row in enumerate(rows):
batch.append(Product(
name=row['name'],
sku=row['sku'],
price=row['price'],
category=row.get('category', 'uncategorized'),
))
if len(batch) >= 500:
Product.objects.bulk_create(batch, ignore_conflicts=True)
batch = []
# Update progress
job.progress = int((i + 1) / total * 100)
job.save(update_fields=['progress'])
# Final batch
if batch:
Product.objects.bulk_create(batch, ignore_conflicts=True)
job.status = 'completed'
job.progress = 100
job.save()
async def start_import(request):
"""Async view that kicks off the import and returns immediately."""
import_job = await ImportJob.objects.acreate(
user=await request.auser(),
status='queued',
progress=0,
)
# Enqueue background task
import_csv_data.enqueue(import_job.id, request.POST['file_path'])
return JsonResponse({
'job_id': import_job.id,
'status': 'queued',
'progress_url': f'/api/imports/{import_job.id}/status/',
})
async def import_status(request, job_id):
"""SSE endpoint for real-time progress updates."""
async def progress_stream():
while True:
job = await ImportJob.objects.aget(pk=job_id)
yield f"data: {json.dumps({'status': job.status, 'progress': job.progress})}\n\n"
if job.status in ('completed', 'failed'):
break
await asyncio.sleep(1)
return StreamingHttpResponse(
progress_stream(),
content_type='text/event-stream'
)
Pattern 3: Async Rate-Limited External API Client
When calling rate-limited external APIs from async views:
import asyncio
class AsyncRateLimiter:
"""Token bucket rate limiter for async contexts."""
def __init__(self, rate_per_second):
self.rate = rate_per_second
self.tokens = rate_per_second
self.last_refill = asyncio.get_event_loop().time()
self._lock = asyncio.Lock()
async def acquire(self):
async with self._lock:
now = asyncio.get_event_loop().time()
elapsed = now - self.last_refill
self.tokens = min(
self.rate,
self.tokens + elapsed * self.rate
)
self.last_refill = now
if self.tokens >= 1:
self.tokens -= 1
return
else:
wait_time = (1 - self.tokens) / self.rate
await asyncio.sleep(wait_time)
self.tokens = 0
# Global rate limiter โ 10 requests per second to external API
api_limiter = AsyncRateLimiter(rate_per_second=10)
async def fetch_enrichment_data(request, entity_id):
await api_limiter.acquire()
async with httpx.AsyncClient() as client:
response = await client.get(
f'https://api.enrichment-service.com/v2/entities/{entity_id}',
headers={'Authorization': f'Bearer {settings.ENRICHMENT_API_KEY}'}
)
return JsonResponse(response.json())
Pattern 4: Async Health Check with Dependency Verification
A production health check that verifies all dependencies concurrently:
import asyncio
import httpx
from django.http import JsonResponse
from django.db import connection
async def health_check(request):
"""
Comprehensive health check that tests all dependencies concurrently.
Returns 200 if all critical services are healthy, 503 otherwise.
"""
async def check_database():
try:
await User.objects.acount()
return {'status': 'healthy', 'latency_ms': None}
except Exception as e:
return {'status': 'unhealthy', 'error': str(e)}
async def check_cache():
try:
from django.core.cache import cache
await cache.aset('health_check', 'ok', timeout=10)
value = await cache.aget('health_check')
return {
'status': 'healthy' if value == 'ok' else 'degraded'
}
except Exception as e:
return {'status': 'unhealthy', 'error': str(e)}
async def check_external_api(name, url):
try:
async with httpx.AsyncClient(timeout=3.0) as client:
resp = await client.get(url)
return {
'status': 'healthy' if resp.status_code == 200 else 'degraded',
'http_status': resp.status_code,
}
except Exception as e:
return {'status': 'unhealthy', 'error': str(e)}
db, cache_result, auth_api, search_api = await asyncio.gather(
check_database(),
check_cache(),
check_external_api('auth', 'https://auth.example.com/health'),
check_external_api('search', 'https://search.example.com/health'),
)
checks = {
'database': db,
'cache': cache_result,
'auth_service': auth_api,
'search_service': search_api,
}
all_healthy = all(
c.get('status') == 'healthy' for c in checks.values()
)
return JsonResponse(
{'status': 'healthy' if all_healthy else 'degraded', 'checks': checks},
status=200 if all_healthy else 503,
)
These patterns share a common theme: async Django's strength is concurrent I/O. Whether you're checking health endpoints, fetching from microservices, streaming AI responses, or monitoring import progress โ the value comes from doing multiple things at once without blocking threads.
19. The Async Decision Framework
After covering every async capability Django offers, the benchmarks, the patterns, and the production lessons, let's synthesize everything into a practical decision framework.
When Async Django Is Worth the Complexity
Use async when you have measurable I/O bottlenecks. The key word is "measurable." Profile your application first. If your views spend most of their time waiting for external API responses, querying multiple independent database tables, or streaming data from third-party services, async will provide concrete, testable improvements.
Specific indicators that async will help:
- Views that call 2 or more external APIs sequentially
- Dashboard views that run 3 or more independent database queries
- Views that integrate with LLM/AI inference APIs (latency-sensitive, often concurrent)
- Real-time features requiring WebSockets or Server-Sent Events
- Streaming responses (file downloads, AI token streaming, log tailing)
When to Stay Synchronous
Stay sync when your bottleneck is the database, not your Python code. The benchmarks showed that sync Django with connection pooling handles 1,822 RPS for database reads โ 3.4x faster than async Django. For standard CRUD operations where each view makes 1-3 database queries, the overhead of the async event loop and sync_to_async wrappers actually makes things slower.
Specific indicators that sync is the better choice:
- Standard CRUD views with simple database queries
- Applications where database query optimization would yield bigger gains
- Teams without async Python expertise (the learning curve is real)
- Applications that are CPU-bound rather than I/O-bound
- Codebases with heavy reliance on third-party packages that aren't async-compatible
The Incremental Path
The beauty of Django's async implementation is that you never have to make an all-or-nothing decision. The recommended approach:
- Start with sync Django and connection pooling โ this alone may solve your performance issues
- Profile to find I/O bottlenecks โ use Django Debug Toolbar, New Relic, or Datadog to identify slow views
- Convert the 3-5 highest-impact views to async โ focus on views with concurrent external calls
- Deploy ASGI alongside WSGI โ route async views to ASGI workers, everything else to WSGI
- Measure the impact โ compare p50, p95, and p99 latencies before and after
- Expand or retreat based on data โ not on hype
Recommended Approach
Incremental
Convert 3-5 high-impact views first
A Note on the Async Criticism
Kevin Renskers at Loopwerk published a widely-discussed article arguing that async Django is "a solution in search of a problem." His core argument: for 99% of web applications, the benefits are marginal, and the framework now requires maintaining parallel APIs (save() vs asave()) that add permanent complexity.
There's truth in this critique. Most Django applications don't need async. The 14% adoption rate after five years of availability supports this. But the 14% who do use it โ teams building LLM interfaces, real-time dashboards, API aggregation layers, and AI-powered tools โ find it genuinely transformative.
The question isn't whether async Django is universally necessary. It's whether your specific application has the I/O characteristics that make it worthwhile. This guide has given you the tools to answer that question with data instead of dogma.
Conclusion
Django's async journey represents one of the most thoughtful framework evolutions in web development history. Rather than chasing the async hype cycle, the Django core team made deliberate choices: maintain backward compatibility, ship incrementally, and let the synchronous path remain a first-class citizen.
Seven years after Andrew Godwin's initial proposal, the result is a framework where every major subsystem โ views, middleware, ORM, authentication, sessions, cache, signals, and pagination โ supports async operation. Django 6.0's background tasks framework and AsyncPaginator represent the latest additions to an async surface area that covers the vast majority of web application needs.
But the most important insight from this guide isn't about async capabilities โ it's about making informed decisions:
-
Sync Django with connection pooling outperforms async Django for straightforward database operations at 1,822 RPS versus 541 RPS. For most applications, enabling psycopg 3's native pooling is the single highest-impact performance optimization available.
-
Async Django excels at concurrent I/O โ the asyncio.gather() pattern for parallel API calls, database queries, and external service integration provides genuine 2-4x latency reductions that justify the added complexity.
-
Real-time capabilities through async streaming responses, Server-Sent Events, and Django Channels give Django superpowers that were impossible in the sync-only era. Streaming AI responses, live dashboards, and WebSocket-based collaboration are all production-ready.
-
The developer experience is improving but not seamless. Kraken Technologies' candid assessment โ that async Django is "okay" โ reflects reality. Foreign key prefetching requirements, transaction wrapper verbosity, and testing friction are real pain points that the Django community continues to address.
-
Production observability changes with async. Event loop monitoring, connection pool tracking, and coroutine-aware logging become essential. Traditional thread-based monitoring tools need adaptation for the event-loop execution model.
The 2024 Django Developer Survey finding โ that developers needing async reach for FastAPI instead of Django's own async โ reflects a perception gap more than a capability gap. Django async is production-ready, battle-tested at companies like Kraken Technologies, and maturing with each release.
As AI coding tools continue to accelerate developer productivity, the barrier to adopting async patterns in existing Django codebases will continue to drop. The framework is ready. The question is whether your use case benefits from it โ and after reading this guide, you should know the answer.
For teams ready to start their Django journey, our Django 6 getting started tutorial covers the fundamentals. For those ready to push Django to its limits in production, the patterns and practices in this guide provide the foundation for building async Django applications that are fast, resilient, and maintainable.
References and Further Reading:
- DEP 0009: Async-capable Django โ Andrew Godwin's original proposal
- Django 6.0 Async Documentation โ Official Django async guide
- Django 6.0 Release Notes โ What's new in the latest release
- Kraken Engineering: Working with async Django โ Production lessons from Kraken Technologies
- Hackeryarn: Async Python Benchmarks โ Rigorous benchmark comparison
- Loopwerk: Async Django โ A Solution in Search of a Problem? โ Critical analysis of async adoption
- JetBrains Django Developer Survey 2024 โ Community usage data
- Django Channels Documentation โ WebSocket and real-time support
- Fly.io: Running Tasks Concurrently in Django โ Practical async patterns
- Saurabh Kumar: Connection Pooling Benchmarks โ PostgreSQL pooling performance data
- Andrew Godwin's Async Roadmap โ The blog post that started it all

