Writing a minimal async Django App

I'm currently working on Dok, a modern documentation viewer with support for fast full text search. I'm not quite there yet but I thought it might be a good idea to write about the process. About the things that have worked out and the ones which turned out to be a dead end. All while learning something new while doing it.

Today, I'm going to build a minimal, modern Django app with async support and websockets.

Modern Django

Django first stable release was in 2005. A lot has changed since then: Progressive web apps have taken over the web, websockets came along, async frameworks like nodejs gained popularity and the heaviest objects in the universe were discovered. Despite all that — and all snakiness aside — Django managed to stay relevant. The reasons for this are many and I don"t want to elaborate on all of them here, but I think one of the main reasons is that Django managed to stay reasonably backwards compatible while continually improving.

Django — and Python in general — have gained a lot of new features in recent years. Writing async code is now part of Pythons standard library and thanks to Channels, Django now also supports websockets. On top of that Django itself is on its way to get full async support.

Prerequisites

First things first, I'm going to install a couple of dependencies, namely Django, channels and uvicorn. I'm using a using a plain old requirements.txt file which I'm going to install into a virtual environment using pip install -r requirements.txt.

django==3.0.6
channels==2.4.0 # websockets for Django
uvicorn==0.11.5 # minimal asgi server

A minimal Django application

I was always amazed by Flasks minimal application example. Using just 5 lines of code spawns a fully working Python powered webserver to play with.

For the minimal Django application, I'm going to create something similar — albeit with a couple more lines of code. My goal is to create an app.py that both configures and runs Django at the same time.

Running Django then becomes as simple as running python app.py.

Here's minimal Django application with full async and Channels support:

import os
# First things first, we are going to set Django's settings module environment variable.
# when the file is being run as `python app.py`, __name__ resolves to "__main__"
os.environ.setdefault("DJANGO_SETTINGS_MODULE", __name__)

# set the debug mode and the secret key
DEBUG = False
SECRET_KEY = "<set me>"

# Configure the urls (in urls.py) and the websocket routing (in routing.py) to be used
# by Django.
ROOT_URLCONF = "urls"
ASGI_APPLICATION = "routing.application"

# We are going to need to configure a couple of basic apps
INSTALLED_APPS = [
    "django.contrib.auth",
    "django.contrib.contenttypes",
    "django.contrib.sessions",
    "django.contrib.messages",
    "channels",
]

# Set the default channel layers backend to the InMemoryChannelLayer
CHANNEL_LAYERS = {
    "default": {
        "BACKEND": "channels.layers.InMemoryChannelLayer"
    }
}


def boot():
    import uvicorn
    import django
    django.setup()
    uvicorn.run(
        "routing:application",
        host="127.0.0.1",
        port=9000,
        log_level="info",
        debug=DEBUG
    )


boot()

Adding urls and websockets

urls.py

from django.urls import path

urlpatterns = [
    path("", lambda s: s),
]

routing.py

from channels.auth import AuthMiddlewareStack
from channels.routing import ProtocolTypeRouter, URLRouter
from channels.security.websocket import AllowedHostsOriginValidator
from django.urls import re_path

from consumers import Consumer

websocket_urlpatterns = [
    re_path(r"$", Consumer),
]

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

Adding a web socket consumer

consumers.py

from channels.generic.websocket import AsyncJsonWebsocketConsumer


class Consumer(AsyncJsonWebsocketConsumer):

    async def connect(self):
        print("connected to consumer")
        await self.channel_layer.group_add(
            f"consumer_group",
            self.channel_name
        )
        await self.accept()

    async def disconnect(self, close_code):
        await self.channel_layer.group_discard(
            "consumer_group",
            self.channel_name
        )
        await self.close()