Django Cheatsheet 10 — Django Async Views

import asyncio
from django.http import JsonResponse

async def view(request):
    await asyncio.sleep(0.1)
    return JsonResponse({"ok": True})

from django.views import View

class MyView(View):
    async def get(self, request):
        return JsonResponse({"ok": True})

uv add uvicorn
uv run uvicorn config.asgi:application --reload

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

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

async def view(request):
    posts = []
    async for post in Post.objects.all():
        posts.append(post.title)
    return JsonResponse({"posts": posts})

# Or:
posts = await Post.objects.all().aexists()
post = await Post.objects.aget(id=1)
count = await Post.objects.acount()
posts = [p async for p in Post.objects.filter(published=True)]
await Post.objects.acreate(title="x")
await post.asave()
await post.adelete()

from asgiref.sync import sync_to_async

async def view(request):
    posts = await sync_to_async(list)(Post.objects.all())
    return JsonResponse({"count": len(posts)})

from asgiref.sync import async_to_sync

def sync_view(request):
    result = async_to_sync(async_func)(arg)

import httpx

async def fetch_view(request):
    async with httpx.AsyncClient() as client:
        r = await client.get("https://api.example.com/x")
    return JsonResponse(r.json())

import asyncio

async def view(request):
    async with httpx.AsyncClient() as client:
        users, posts = await asyncio.gather(
            client.get("https://api/users"),
            client.get("https://api/posts"),
        )
    return JsonResponse({"users": users.json(), "posts": posts.json()})

from django.http import StreamingHttpResponse
import asyncio

async def gen():
    for i in range(10):
        yield f"event: {i}\n"
        await asyncio.sleep(1)

async def stream_view(request):
    return StreamingHttpResponse(gen(), content_type="text/plain")

async def sse(request):
    async def stream():
        while True:
            data = await fetch_data()
            yield f"data: {data}\n\n"
            await asyncio.sleep(2)
    return StreamingHttpResponse(stream(), content_type="text/event-stream")

class AsyncMiddleware:
    sync_capable = False
    async_capable = True
    
    def __init__(self, get_response):
        self.get_response = get_response
    
    async def __call__(self, request):
        # Before
        response = await self.get_response(request)
        # After
        return response

from django.dispatch import receiver

@receiver(post_save, sender=Post)
async def handle(sender, instance, **kwargs):
    await notify(instance.id)

from django.core.cache import cache

await cache.aset("key", value, timeout=60)
value = await cache.aget("key")
await cache.adelete("key")

# BAD: sync ORM in async view
async def view(request):
    posts = Post.objects.all()        # SynchronousOnlyOperation
    return ...

# GOOD: async ORM
async def view(request):
    posts = [p async for p in Post.objects.all()]

uv add channels channels-redis

INSTALLED_APPS = [..., "channels"]
ASGI_APPLICATION = "config.asgi.application"
CHANNEL_LAYERS = {
    "default": {
        "BACKEND": "channels_redis.core.RedisChannelLayer",
        "CONFIG": {"hosts": [("localhost", 6379)]},
    },
}

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

class ChatConsumer(AsyncWebsocketConsumer):
    async def connect(self):
        await self.channel_layer.group_add("chat", self.channel_name)
        await self.accept()
    
    async def disconnect(self, code):
        await self.channel_layer.group_discard("chat", self.channel_name)
    
    async def receive(self, text_data):
        await self.channel_layer.group_send("chat", {
            "type": "chat.message",
            "message": text_data,
        })
    
    async def chat_message(self, event):
        await self.send(text_data=event["message"])

# routing.py
from django.urls import path
from .consumers import ChatConsumer

websocket_urlpatterns = [
    path("ws/chat/", ChatConsumer.as_asgi()),
]

# config/asgi.py
from channels.routing import ProtocolTypeRouter, URLRouter
from channels.auth import AuthMiddlewareStack
from blog.routing import websocket_urlpatterns

application = ProtocolTypeRouter({
    "http": get_asgi_application(),
    "websocket": AuthMiddlewareStack(URLRouter(websocket_urlpatterns)),
})