Async Support

import os

from django.core.asgi import get_asgi_application

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "config.settings")

application = get_asgi_application()

import asyncio
import threading

from django.http import HttpResponse, JsonResponse
from django.shortcuts import aget_object_or_404

from .models import Author, Book
from .services import fetch_rating


async def hello(request):
    await asyncio.sleep(0.1)  # stands in for an awaited network call
    return HttpResponse("Hello, async world!")


def where_am_i(request):
    """Works for sync and async views: report the thread and whether it has a loop."""
    try:
        asyncio.get_running_loop()
        has_loop = True
    except RuntimeError:
        has_loop = False
    return JsonResponse({"thread": threading.current_thread().name, "loop_in_thread": has_loop})


async def where_am_i_async(request):
    return where_am_i(request)

import asyncio

from django.http import JsonResponse
from django.utils.decorators import method_decorator
from django.views import View
from django.views.decorators.cache import never_cache
from django.views.generic import TemplateView

from .models import Author, Book


class BookStatsView(View):
    """Every HTTP handler is `async def`, so as_view() returns a coroutine function."""

    async def get(self, request, *args, **kwargs):
        books, authors = await asyncio.gather(Book.objects.acount(), Author.objects.acount())
        return JsonResponse({"books": books, "authors": authors})

    async def post(self, request, *args, **kwargs):
        book = await Book.objects.acreate(
            title=request.POST["title"], author=await Author.objects.afirst()
        )
        return JsonResponse({"id": book.pk}, status=201)


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

    def post(self, request):  # a sync handler next to an async one
        return JsonResponse({"ok": True})


class NeverCachedStatsView(BookStatsView):
    # Decorating dispatch() needs an async dispatch(), or method_decorator()
    # cannot tell that the result is a coroutine function.
    @method_decorator(never_cache)
    async def dispatch(self, *args, **kwargs):
        return await super().dispatch(*args, **kwargs)


class TitlesPage(TemplateView):
    template_name = "catalog/titles.html"

    async def get(self, request, *args, **kwargs):
        context = {"books": Book.objects.all()}  # lazy queryset, on purpose
        return self.render_to_response(context)  # a TemplateResponse

class AsyncCBVTests(AsyncTestBase):
    async def test_async_handlers(self):
        self.assertTrue(inspect.iscoroutinefunction(BookStatsView.as_view()))
        data = (await self.async_client.get("/stats/")).json()
        self.assertEqual(data, {"books": 3, "authors": 2})
        response = await self.async_client.post("/stats/", {"title": "Wild Seed"})
        self.assertEqual(response.status_code, 201)

    async def test_options_and_405_are_wrapped(self):
        response = await self.async_client.options("/stats/")
        self.assertEqual(response["Allow"], "GET, POST, HEAD, OPTIONS")
        self.assertEqual((await self.async_client.delete("/stats/")).status_code, 405)

    def test_mixed_handlers_are_refused_by_as_view(self):
        with self.assertRaisesMessage(ImproperlyConfigured, "must either be all sync or all async"):
            MixedView.as_view()

    async def test_decorated_dispatch(self):
        response = await self.async_client.get("/stats-nocache/")
        self.assertIn("no-cache", response["Cache-Control"])

    async def test_template_response_with_a_lazy_queryset_works(self):
        response = await self.async_client.get("/titles/")
        self.assertEqual(response.content.decode().strip(), "Dawn;Kindred;The Dispossessed;")

    async def test_render_shortcut_with_a_lazy_queryset_fails(self):
        from django.shortcuts import render

        request = AsyncRequestFactory().get("/")
        with self.assertRaises(SynchronousOnlyOperation):
            render(request, "catalog/titles.html", {"books": Book.objects.all()})

import time
from inspect import iscoroutinefunction

from django.utils.decorators import sync_and_async_middleware


@sync_and_async_middleware
def server_timing(get_response):
    """Adds a Server-Timing header; works whichever way Django calls it."""
    if iscoroutinefunction(get_response):

        async def middleware(request):
            start = time.perf_counter()
            response = await get_response(request)
            response["Server-Timing"] = f"app;dur={(time.perf_counter() - start) * 1000:.1f}"
            return response

    else:

        def middleware(request):
            start = time.perf_counter()
            response = get_response(request)
            response["Server-Timing"] = f"app;dur={(time.perf_counter() - start) * 1000:.1f}"
            return response

    return middleware

import logging
from inspect import iscoroutinefunction, markcoroutinefunction

log = logging.getLogger("catalog")


class RequestIdMiddleware:
    """Async-only class-based middleware."""

    sync_capable = False
    async_capable = True

    def __init__(self, get_response):
        self.get_response = get_response
        if iscoroutinefunction(self.get_response):
            markcoroutinefunction(self)

    async def __call__(self, request):
        request.request_id = f"req-{id(request) % 100000}"
        response = await self.get_response(request)
        response["X-Request-Id"] = request.request_id
        return response

    async def process_view(self, request, view_func, view_args, view_kwargs):
        log.info("process_view %s", view_func.__name__)
        return None


class LegacySyncMiddleware:
    """A plain synchronous middleware, the way most third-party packages ship."""

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

    def __call__(self, request):
        response = self.get_response(request)
        response["X-Legacy"] = "sync"
        return response

from django.db import models


class Author(models.Model):
    name = models.CharField(max_length=200)

    def __str__(self):
        return self.name


class Book(models.Model):
    title = models.CharField(max_length=200)
    author = models.ForeignKey(Author, related_name="books", on_delete=models.CASCADE)
    pages = models.PositiveIntegerField(default=0)
    summary = models.TextField(blank=True)

    class Meta:
        ordering = ["title"]

    def __str__(self):
        return self.title

async def book_list(request):
    books = [
        {"id": b.pk, "title": b.title, "author": b.author.name}
        async for b in Book.objects.select_related("author")
    ]
    return JsonResponse({"count": await Book.objects.acount(), "books": books})


async def book_detail(request, pk):
    book = await aget_object_or_404(Book.objects.select_related("author"), pk=pk)
    return JsonResponse({"id": book.pk, "title": book.title, "author": book.author.name})


async def book_create(request):
    """POST title and author name; shows acreate, aget_or_create and asave."""
    author, created = await Author.objects.aget_or_create(name=request.POST["author"])
    book = await Book.objects.acreate(title=request.POST["title"], author=author)
    book.pages = int(request.POST.get("pages", 0))
    await book.asave(update_fields=["pages"])
    return JsonResponse({"id": book.pk, "author_created": created}, status=201)

import asyncio
import threading

from asgiref.sync import async_to_sync, sync_to_async

from .models import Book
from .services import fetch_rating


def thread_name():
    return threading.current_thread().name


def sync_work(pk):
    """Ordinary blocking code: ORM, signals, a legacy library."""
    return Book.objects.get(pk=pk).title


async def adapter_demo():
    in_loop = thread_name()
    sensitive = await asyncio.gather(*(sync_to_async(thread_name)() for _ in range(5)))
    parallel = await asyncio.gather(
        *(sync_to_async(thread_name, thread_sensitive=False)() for _ in range(5))
    )
    title = await sync_to_async(sync_work)(1)
    return {"loop": in_loop, "sensitive": sorted(set(sensitive)),
            "not_sensitive": sorted(set(parallel)), "title": title}


@sync_to_async
def decorated_helper(pk):
    return sync_work(pk).upper()


def call_async_from_sync(book_id):
    """In a synchronous view or a management command."""
    return async_to_sync(fetch_rating)(book_id)


async def wrong_direction():
    return async_to_sync(fetch_rating)(1)  # sync-over-async inside the event loop

$ python run_adapters.py
{'loop': 'MainThread', 'sensitive': ['ThreadPoolExecutor-0_0'], 'not_sensitive': ['asyncio_0', 'asyncio_1'], 'title': 'The Dispossessed'}
THE DISPOSSESSED
RuntimeError You cannot use AsyncToSync in the same thread as an async event loop - just await the async function directly.
{'book': 3, 'rating': 4.3}

import threading

from asgiref.sync import async_to_sync


async def async_only_library_call(book_id):
    rating = await fetch_rating(book_id)
    return {**rating, "coroutine_ran_in": threading.current_thread().name}


def sync_view_calling_async(request, pk):
    """A plain `def` view that needs an async-only library."""
    data = async_to_sync(async_only_library_call)(pk)
    data["view_ran_in"] = threading.current_thread().name
    return JsonResponse(data)

$ python -m uvicorn config.asgi:application --port 8759 &
$ curl -s http://127.0.0.1:8759/sync-calls-async/3/
{"book": 3, "rating": 4.3, "coroutine_ran_in": "MainThread", "view_ran_in": "ThreadPoolExecutor-1_0"}

from django.db import connection, transaction
from django.template import Context, Template

from .models import Book


async def sync_get():
    return Book.objects.get(pk=1)  # sync ORM call from async code


async def sync_iteration():
    return [b.title for b in Book.objects.all()]  # iteration runs the query


async def sync_len():
    return len(Book.objects.all())  # so does len(), list(), bool()


async def lazy_foreign_key():
    book = await Book.objects.aget(pk=1)
    return book.author.name  # the author is fetched lazily, synchronously


async def deferred_field():
    book = await Book.objects.defer("summary").aget(pk=1)
    return book.summary  # a deferred field loads lazily, synchronously


async def atomic_block():
    with transaction.atomic():  # transactions are not async-capable
        await Book.objects.acreate(title="Lost", author_id=1)


async def raw_cursor():
    with connection.cursor() as cursor:  # the connection is async-unsafe
        cursor.execute("SELECT 1")


async def template_loop():
    template = Template("{% for b in books %}{{ b.title }};{% endfor %}")
    return template.render(Context({"books": Book.objects.all()}))

$ python run_unsafe.py
sync_get -> django.core.exceptions.SynchronousOnlyOperation: You cannot call this from an async context - use a thread or sync_to_async.
sync_iteration -> django.core.exceptions.SynchronousOnlyOperation: You cannot call this from an async context - use a thread or sync_to_async.
...
template_loop -> django.core.exceptions.SynchronousOnlyOperation: You cannot call this from an async context - use a thread or sync_to_async.

async def broken(request):
    title = Book.objects.get(pk=1).title  # synchronous ORM call inside async def
    return HttpResponse(title)

import logging

from django.dispatch import Signal, receiver

log = logging.getLogger("catalog")

book_viewed = Signal()


@receiver(book_viewed)
def count_view_sync(sender, book_id, **kwargs):
    log.info("sync receiver saw book %s", book_id)
    return "sync"


@receiver(book_viewed)
async def count_view_async(sender, book_id, **kwargs):
    log.info("async receiver saw book %s", book_id)
    return "async"

class AsyncServicesTests(AsyncTestBase):
    async def test_login_whoami_logout_and_permissions(self):
        password = secrets.token_urlsafe(16)
        user = await User.objects.acreate_user("ada", password=password)  # 5.2
        client = AsyncClient()
        self.assertFalse((await client.get("/whoami/")).json()["authenticated"])
        bad = await client.post("/login/", {"username": "ada", "password": "wrong"})
        self.assertEqual(bad.status_code, 400)
        ok = await client.post("/login/", {"username": "ada", "password": password})
        self.assertEqual(ok.json(), {"user": "ada"})
        self.assertEqual((await client.get("/whoami/")).json()["user"], "ada")
        self.assertFalse((await client.get("/can-add/")).json()["can_add"])
        perm = await Permission.objects.aget(codename="add_book")
        await user.user_permissions.aadd(perm)  # related-manager async method, 4.2
        user = await User.objects.aget(pk=user.pk)  # drop the cached permissions
        self.assertTrue(await user.ahas_perm("catalog.add_book"))
        await client.post("/logout/")
        self.assertEqual((await client.get("/can-add/")).status_code, 302)  # login_required

    async def test_alogin_helper_of_the_test_client(self):
        user = await User.objects.acreate_user("grace", password=secrets.token_urlsafe(16))
        client = AsyncClient()
        await client.aforce_login(user)  # 5.0
        self.assertEqual((await client.get("/whoami/")).json()["user"], "grace")

    async def test_request_user_is_lazy_and_synchronous(self):
        user = await User.objects.acreate_user("alan", password=secrets.token_urlsafe(16))
        client = AsyncClient()
        await client.aforce_login(user)
        with self.assertRaises(SynchronousOnlyOperation):
            await client.get("/whoami-wrong/")

    async def test_sessions(self):
        client = AsyncClient()
        self.assertEqual((await client.get("/visits/")).json(), {"visits": 1})
        self.assertEqual((await client.get("/visits/")).json(), {"visits": 2})
        session = await client.asession()  # 5.0
        self.assertEqual(await session.aget("visits"), 2)

    async def test_cache(self):
        await cache.aclear()
        self.assertFalse((await self.async_client.get("/rating/4/")).json()["hit"])
        self.assertTrue((await self.async_client.get("/rating/4/")).json()["hit"])

    async def test_signals_sync_and_async_receivers(self):
        with self.assertLogs("catalog", level="INFO") as logs:
            data = (await self.async_client.get(f"/viewed/{self.disp.pk}/")).json()
        self.assertEqual(data["receivers"], ["async", "sync"])
        self.assertEqual(len(logs.output), 2)
        robust = await book_viewed.asend_robust(sender=Book, book_id=1)
        self.assertEqual(len(robust), 2)

    async def test_async_paginator(self):
        page1 = (await self.async_client.get("/paged/")).json()
        self.assertEqual(page1, {"page": 1, "pages": 2, "has_next": True, "titles": ["Dawn", "Kindred"]})
        page2 = (await self.async_client.get("/paged/?page=2")).json()
        self.assertEqual(page2["titles"], ["The Dispossessed"])
        self.assertFalse(page2["has_next"])

    async def test_aenqueue(self):
        data = (await self.async_client.get(f"/reindex/{self.disp.pk}/")).json()
        self.assertEqual(data, {"status": "SUCCESSFUL", "value": f"indexed {self.disp.pk}"})

import asyncio
import json
import logging

from django.http import StreamingHttpResponse

log = logging.getLogger("catalog")


async def book_feed(request):
    """Newline-delimited JSON, one book at a time, from an async generator."""

    async def rows():
        try:
            async for book in Book.objects.order_by("title").values("id", "title").aiterator(chunk_size=100):
                yield json.dumps(book) + "\n"
                await asyncio.sleep(0)  # a real feed would await new data here
        except asyncio.CancelledError:
            log.info("feed client went away")
            raise

    return StreamingHttpResponse(rows(), content_type="application/x-ndjson")

$ curl -s -N http://127.0.0.1:8759/feed/ | head -1
{"id": 3, "title": "Dawn"}

paginator = AsyncPaginator(Book.objects.order_by("title"), per_page=2)
page = await paginator.aget_page(request.GET.get("page"))
titles = [b.title for b in await page.aget_object_list()]

$ python manage.py test catalog
Found 43 test(s).
System check identified no issues (0 silenced).
...........................................
Ran 43 tests in 3.185s

OK