James Tech Blog
Published at

Django Ninja and Async Views

Django Ninja and Async Views

Explore Django Ninja's async views for efficient concurrent operations. Learn how to implement and test async views, mix sync/async code, and use async ORM features.

Authors
  • avatar
    Name
    James Lau
    Twitter
  • Indie App Developer at Self-employed
Sharing is caring!
Table of Contents

Since version 3.1, Django has provided support for asynchronous views. This feature enables you to create efficient, concurrent views, particularly beneficial for network-bound and I/O-bound operations. Django Ninja fully supports async views, making it straightforward to implement them in your projects.

Benefits of Async Views

Async views can significantly improve performance in scenarios such as:

  • Calling external APIs over the network
  • Executing and waiting for database queries
  • Reading from and writing to disk drives

Quick Example

Let’s illustrate with an example. Suppose we have an API endpoint that performs some work (in this case, sleeps for a specified duration) and returns a word:

import time

@api.get("/say-after")
def say_after(request, delay: int, word: str):
    time.sleep(delay)
    return {"saying": word}

To convert this to an asynchronous view, simply add the async keyword to the function definition. Also, replace blocking calls (like time.sleep) with their async equivalents (like asyncio.sleep):

import asyncio

@api.get("/say-after")
async def say_after(request, delay: int, word: str):
    await asyncio.sleep(delay)
    return {"saying": word}

Running Async Views

To run async views, you need an ASGI server such as Uvicorn or Daphne. Here’s how to use Uvicorn:

First, install Uvicorn:

pip install uvicorn

Then, start the server:

uvicorn your_project.asgi:application --reload

Note: Replace your_project with your actual project package name. The --reload flag enables automatic server reloading on code changes, which is useful for development but should be avoided in production.

Important: While you can run async views with manage.py runserver, it’s generally recommended to use an ASGI server like Uvicorn or Daphne, especially for libraries that benefit from a fully async environment.

Testing Async Views

Visit your endpoint in the browser, e.g., http://127.0.0.1:8000/api/say-after?delay=3&word=hello. After a 3-second delay, you should see the response {"saying": "hello"}.

To test concurrency, you can use a tool like ab (ApacheBench) to send multiple parallel requests:

ab -c 100 -n 100 "http://127.0.0.1:8000/api/say-after?delay=3&word=hello"

This command sends 100 concurrent requests. The results will show how efficiently your service handles these requests.

Mixing Sync and Async Operations

You can seamlessly mix synchronous and asynchronous views in your Django Ninja project. Django Ninja intelligently routes requests to the appropriate handler:

import time
import asyncio

@api.get("/say-sync")
def say_after_sync(request, delay: int, word: str):
    time.sleep(delay)
    return {"saying": word}

@api.get("/say-async")
async def say_after_async(request, delay: int, word: str):
    await asyncio.sleep(delay)
    return {"saying": word}

Real-World Example: Elasticsearch

Let’s consider a practical example using Elasticsearch, which now offers async support. First, install the Elasticsearch library:

pip install elasticsearch>=7.8.0

Then, use the AsyncElasticsearch class and await the results:

from ninja import NinjaAPI
from elasticsearch import AsyncElasticsearch

api = NinjaAPI()

es = AsyncElasticsearch()

@api.get("/search")
async def search(request, q: str):
    resp = await es.search(
        index="documents",
        body={"query": {"query_string": {"query": q}}},
        size=20,
    )
    return resp["hits"]

Working with the ORM

Currently, Django’s ORM has limitations in async environments due to global state that isn’t coroutine-aware. These parts are considered “async-unsafe”.

Attempting to directly use the ORM in an async view like this:

@api.get("/blog/{post_id}")
async def search(request, post_id: int):
    blog = Blog.objects.get(pk=post_id)
    ...

will raise an error. To work around this, use the sync_to_async() adapter:

from asgiref.sync import sync_to_async

@sync_to_async
def get_blog(post_id):
    return Blog.objects.get(pk=post_id)

@api.get("/blog/{post_id}")
async def search(request, post_id: int):
    blog = await get_blog(post_id)
    ...

Or, more concisely:

@api.get("/blog/{post_id}")
async def search(request, post_id: int):
    blog = await sync_to_async(Blog.objects.get)(pk=post_id)
    ...

Important GOTCHA: Django querysets are lazily evaluated. Therefore, the following will NOT work:

all_blogs = await sync_to_async(Blog.objects.all)()
# it will throw an error later when you try to iterate over all_blogs
...

Instead, force evaluation, for example, by converting to a list:

all_blogs = await sync_to_async(list)(Blog.objects.all())
...

Async ORM Operations (Django 4.1+)

Since Django 4.1, asynchronous versions of ORM operations are available, eliminating the need for sync_to_async in many cases. These async operations have the same names as their synchronous counterparts but are prefixed with a. For example:

@api.get("/blog/{post_id}")
async def search(request, post_id: int):
    blog = await Blog.objects.aget(pk=post_id)
    ...

When working with querysets, use async for paired with list comprehension:

all_blogs = [blog async for blog in Blog.objects.all()]
...
Sharing is caring!