Three years ago I wrote an article about webockets. In fact I’ve written several articles about Websockets (Websockets and real time communications is something that I’m really passionate about), but today I would like to pick up this article. Nowadays I’m involved with several Django projects so I want to create a similar working prototype with Django. Let’s start:

In the past I normally worked with libraries such as socket.io to ensure browser compatibility with Websockets. Nowadays, at least in my world, we can assume that our users are using a modern browser with websocket support, so we’re going to use plain Websockets instead external libraries. Django has a great support to Websockets called Django Channels. It allows us to to handle Websockets (and other async protocols) thanks to Python’s ASGI’s specification. In fact is pretty straightforward to build applications with real time communication and with shared authentication (something that I have done in the past with a lot of effort. I’m getting old and now I like simple things :))

The application that I want to build is the following one: One Web application that shows the current time with seconds. Ok it’s very simple to do it with a couple of javascript lines but this time I want to create a worker that emits an event via Websockets with the current time. My web application will show that real time update. This kind of architecture always have the same problem: The initial state. In this example we can ignore it. When the user opens the browser it must show the current time. As I said before in this example we can ignore this situation. We can wait until the next event to update the initial blank information but if the event arrives each 10 seconds our user will have a blank screen until the next event arrives. In our example we’re going to take into account this situation. Each time our user connects to the Websocket it will ask to the server for the initial state.

Our initial state route will return the current time (using Redis). We can authorize our route using the standard Django’s protected routes

[sourcecode language=”python”]
from django.contrib.auth.decorators import login_required
from django.http import JsonResponse
from ws.redis import redis

@login_required
def initial_state(request):
return JsonResponse({‘current’: redis.get(‘time’)})
[/sourcecode]

We need to configure our channels and define a our event:

[sourcecode language=”python”]
from django.urls import re_path

from ws import consumers

websocket_urlpatterns = [
re_path(r’time/tic/$’, consumers.WsConsumer),
]
[/sourcecode]

As we can see here we can reuse the authentication middleware in channel’s consumers also.
[sourcecode language=”python”]
import json
import json
from channels.generic.websocket import AsyncWebsocketConsumer

class WsConsumer(AsyncWebsocketConsumer):
GROUP = ‘time’

async def connect(self):
if self.scope["user"].is_anonymous:
await self.close()
else:
await self.channel_layer.group_add(
self.GROUP,
self.channel_name
)
await self.accept()

async def tic_message(self, event):
if not self.scope["user"].is_anonymous:
message = event[‘message’]

await self.send(text_data=json.dumps({
‘message’: message
}))
[/sourcecode]

We’re going to need a worker that each second triggers the current time (to avoid problems we’re going to trigger our event each 0.5 seconds). To perform those kind of actions Django has a great tool called Celery. We can create workers and scheduled task with Celery (exactly what we need in our example). To avoid the “initial state” situation our worker will persists the initial state into a Redis storage

[sourcecode language=”python”]
app = Celery(‘config’)
app.config_from_object(‘django.conf:settings’, namespace=’CELERY’)
app.autodiscover_tasks()

@app.on_after_configure.connect
def setup_periodic_tasks(sender, **kwargs):
sender.add_periodic_task(0.5, ws_beat.s(), name=’beat every 0.5 seconds’)

@app.task
def ws_beat(group=WsConsumer.GROUP, event=’tic_message’):
current_time = time.strftime(‘%X’)
redis.set(‘time’, current_time)
message = {‘time’: current_time}
channel_layer = channels.layers.get_channel_layer()
async_to_sync(channel_layer.group_send)(group, {‘type’: event, ‘message’: message})
[/sourcecode]

Finally we need a javascript client to consume our Websockets

[sourcecode language=”javascript”]
let getWsUri = () => {
return window.location.protocol === "https:" ? "wss" : "ws" +
‘://’ + window.location.host +
"/time/tic/"
}

let render = value => {
document.querySelector(‘#display’).innerHTML = value
}

let ws = new ReconnectingWebSocket(getWsUri())

ws.onmessage = e => {
const data = JSON.parse(e.data);
render(data.message.time)
}

ws.onopen = async () => {
let response = await axios.get("/api/initial_state")
render(response.data.current)
}
[/sourcecode]

Basically that’s the source code (plus Django the stuff).

Application architecture
The architecture of the application is the following one:

Frontend
The Django application. We can run this application in development with
python manage.py runserver

And in production using a asgi server (uvicorn in this case)
[sourcecode language=”xml”]
uvicorn config.asgi:application –port 8000 –host 0.0.0.0 –workers 1
[/sourcecode]

In development mode:
[sourcecode language=”xml”]
celery -A ws worker -l debug
[/sourcecode]

And in production
[sourcecode language=”xml”]
celery -A ws worker –uid=nobody –gid=nogroup
[/sourcecode]

We need this scheduler to emit our event (each 0.5 seconds)
[sourcecode language=”xml”]
celery -A ws beat
[/sourcecode]

Message Server for Celery
In this case we’re going to use Redis

Docker
With this application we can use the same dockerfile for frontend, worker and scheduler using different entrypoints

Dockerfile:

[sourcecode language=”xml”]
FROM python:3.8

ENV TZ ‘Europe/Madrid’
RUN echo $TZ > /etc/timezone && \
apt-get update && apt-get install -y tzdata && \
rm /etc/localtime && \
ln -snf /usr/share/zoneinfo/$TZ /etc/localtime && \
dpkg-reconfigure -f noninteractive tzdata && \
apt-get clean

ENV PYTHONDONTWRITEBYTECODE 1
ENV PYTHONUNBUFFERED 1

ADD . /src
WORKDIR /src

RUN pip install -r requirements.txt

RUN mkdir -p /var/run/celery /var/log/celery
RUN chown -R nobody:nogroup /var/run/celery /var/log/celery
[/sourcecode]

And our whole application within a docker-compose file

[sourcecode language=”xml”]
version: ‘3.4’

services:
redis:
image: redis
web:
image: clock:latest
command: /bin/bash ./docker-entrypoint.sh
healthcheck:
test: ["CMD", "curl", "-f", "http://localhost:8000/health"%5D
interval: 1m30s
timeout: 10s
retries: 3
start_period: 40s
depends_on:
– "redis"
ports:
– 8000:8000
environment:
ENVIRONMENT: prod
REDIS_HOST: redis
celery:
image: clock:latest
command: celery -A ws worker –uid=nobody –gid=nogroup
depends_on:
– "redis"
environment:
ENVIRONMENT: prod
REDIS_HOST: redis
cron:
image: clock:latest
command: celery -A ws beat
depends_on:
– "redis"
environment:
ENVIRONMENT: prod
REDIS_HOST: redis
[/sourcecode]

If we want to deploy our application in a K8s cluster we need to migrate our docker-compose file into a k8s yaml files. I assume that we’ve deployed our docker containers into a container registry (such as ECR)

Frontend:
[sourcecode language=”xml”]
apiVersion: apps/v1
kind: Deployment
metadata:
name: clock-web-api
spec:
replicas: 1
selector:
matchLabels:
app: clock-web-api
project: clock
template:
metadata:
labels:
app: clock-web-api
project: clock
spec:
containers:
– name: web-api
image: my-ecr-path/clock:latest
args: ["uvicorn", "config.asgi:application", "–port", "8000", "–host", "0.0.0.0", "–workers", "1"]
ports:
– containerPort: 8000
env:
– name: REDIS_HOST
valueFrom:
configMapKeyRef:
name: clock-app-config
key: redis.host
—
apiVersion: v1
kind: Service
metadata:
name: clock-web-api
spec:
type: LoadBalancer
selector:
app: clock-web-api
project: clock
ports:
– protocol: TCP
port: 8000 # port exposed internally in the cluster
targetPort: 8000 # the container port to send requests to
[/sourcecode]

Celery worker
[sourcecode language=”xml”]
apiVersion: apps/v1
kind: Deployment
metadata:
name: clock-web-api
spec:
replicas: 1
selector:
matchLabels:
app: clock-web-api
project: clock
template:
metadata:
labels:
app: clock-web-api
project: clock
spec:
containers:
– name: web-api
image: my-ecr-path/clock:latest
args: ["uvicorn", "config.asgi:application", "–port", "8000", "–host", "0.0.0.0", "–workers", "1"]
ports:
– containerPort: 8000
env:
– name: REDIS_HOST
valueFrom:
configMapKeyRef:
name: clock-app-config
key: redis.host
—
apiVersion: v1
kind: Service
metadata:
name: clock-web-api
spec:
type: LoadBalancer
selector:
app: clock-web-api
project: clock
ports:
– protocol: TCP
port: 8000 # port exposed internally in the cluster
targetPort: 8000 # the container port to send requests to
[/sourcecode]

Celery scheduler
[sourcecode language=”xml”]
apiVersion: apps/v1
kind: Deployment
metadata:
name: clock-cron
spec:
replicas: 1
selector:
matchLabels:
app: clock-cron
project: clock
template:
metadata:
labels:
app: clock-cron
project: clock
spec:
containers:
– name: clock-cron
image: my-ecr-path/clock:latest
args: ["celery", "-A", "ws", "beat"]
env:
– name: REDIS_HOST
valueFrom:
configMapKeyRef:
name: clock-app-config
key: redis.host
[/sourcecode]

Redis
[sourcecode language=”xml”]
apiVersion: apps/v1
kind: Deployment
metadata:
name: clock-redis
spec:
replicas: 1
selector:
matchLabels:
app: clock-redis
project: clock
template:
metadata:
labels:
app: clock-redis
project: clock
spec:
containers:
– name: clock-redis
image: redis
ports:
– containerPort: 6379
name: clock-redis
—
apiVersion: v1
kind: Service
metadata:
name: clock-redis
spec:
type: ClusterIP
ports:
– port: 6379
targetPort: 6379
selector:
app: clock-redis
[/sourcecode]

Shared configuration
[sourcecode language=”xml”]
apiVersion: v1
kind: ConfigMap
metadata:
name: clock-app-config
data:
redis.host: "clock-redis"
[/sourcecode]

We can deploy or application to our k8s cluster

[sourcecode language=”xml”]
kubectl apply -f .k8s/
[/sourcecode]

And see it running inside the cluster locally with a port forward

[sourcecode language=”xml”]
kubectl port-forward deployment/clock-web-api 8000:8000
[/sourcecode]

And that’s all. Our Django application with Websockets using Django Channels up and running with docker and also using k8s.

Source code in my github