The reliable way to use MQTT with Django is to keep the MQTT connection out of the web process. Run a long-lived subscriber as a custom management command, let it write to the database with the normal ORM, publish from views with a short-lived client, and use Django Channels when browsers need the messages live.
This tutorial builds exactly that with paho-mqtt 2.x, then shows how to deploy the worker and test the whole path end to end.
Why not connect in apps.py or a view?
The first idea most people have is to start a paho client in AppConfig.ready(). It works on your laptop, then breaks in subtle ways:
| Where you start the client | What goes wrong |
|---|---|
AppConfig.ready() | Runs in every process that loads Django: runserver (twice, because of the autoreloader), every Gunicorn worker, migrate, shell, Celery workers. Each opens a connection. |
| Inside a view | A new MQTT session per HTTP request: slow, floods the broker with CONNECTs and fails as soon as two requests share a client ID. |
| A management command | Exactly one process you start on purpose, easy to supervise and restart. |
Several processes subscribing to the same topic means every message is handled several times, and if they share a client ID the broker kicks them off in a loop (see duplicate client ID). So: subscribe in a worker, publish from wherever you need to.
Project setup
pip install django paho-mqtt channels channels-redis daphne
django-admin startproject mysite .
python manage.py startapp telemetryAdd telemetry to INSTALLED_APPS and a model for incoming readings:
from django.db import models
class Reading(models.Model):
device = models.CharField(max_length=64, db_index=True)
topic = models.CharField(max_length=255)
payload = models.JSONField()
received_at = models.DateTimeField(auto_now_add=True)
class Meta:
ordering = ["-received_at"]Keep connection settings in settings.py, read from the environment:
import os
MQTT_HOST = os.environ.get("MQTT_HOST", "broker.connectmq.com")
MQTT_PORT = int(os.environ.get("MQTT_PORT", "8883"))
MQTT_USER = os.environ.get("MQTT_USER", "")
MQTT_PASS = os.environ.get("MQTT_PASS", "")
MQTT_TOPIC = "testmqtt/django/+/telemetry"Run python manage.py makemigrations telemetry && python manage.py migrate.
The MQTT worker as a management command
Create telemetry/management/commands/mqtt_listen.py (both management/ and commands/ need an empty __init__.py):
import json
import os
import socket
import paho.mqtt.client as mqtt
from django.conf import settings
from django.core.management.base import BaseCommand
from django.db import close_old_connections
from telemetry.models import Reading
class Command(BaseCommand):
help = "Subscribe to MQTT telemetry and store it"
def handle(self, *args, **options):
client_id = f"django-worker-{socket.gethostname()}-{os.getpid()}"
client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=client_id)
client.username_pw_set(settings.MQTT_USER, settings.MQTT_PASS)
client.tls_set() # system CA bundle, verifies the hostname
client.reconnect_delay_set(min_delay=1, max_delay=60)
client.on_connect = self.on_connect
client.on_message = self.on_message
client.on_disconnect = self.on_disconnect
client.connect(settings.MQTT_HOST, settings.MQTT_PORT, keepalive=60)
self.stdout.write(f"Connecting as {client_id}")
try:
client.loop_forever() # blocks, reconnects automatically
except KeyboardInterrupt:
client.disconnect()
def on_connect(self, client, userdata, flags, reason_code, properties):
if reason_code.is_failure:
self.stderr.write(f"Connect failed: {reason_code}")
return
# Subscribe here so the subscription is restored after every reconnect
client.subscribe(settings.MQTT_TOPIC, qos=1)
self.stdout.write(f"Subscribed to {settings.MQTT_TOPIC}")
def on_disconnect(self, client, userdata, flags, reason_code, properties):
self.stderr.write(f"Disconnected: {reason_code}")
def on_message(self, client, userdata, msg):
close_old_connections() # drop DB connections older than CONN_MAX_AGE
try:
data = json.loads(msg.payload)
except (UnicodeDecodeError, json.JSONDecodeError):
self.stderr.write(f"Ignoring non-JSON payload on {msg.topic}")
return
device = msg.topic.split("/")[2]
Reading.objects.create(device=device, topic=msg.topic, payload=data)A few details matter here:
- Subscribe in
on_connect, not afterconnect(). paho reconnects on its own, and subscribing in the callback restores the subscription every time. loop_forever()runs callbacks in the main thread, so synchronous ORM calls are safe. Nosync_to_asyncneeded in this process.close_old_connections()is what Django calls at the start and end of each request. A worker never sees a request, so call it yourself or the process will eventually fail with a closed database connection after a database restart.- A unique client ID per process (hostname + PID). Generate a different pattern with the client ID generator if you prefer.
- Never crash on bad payloads. An unhandled exception in a callback stops the loop. For more on parsing and validation, see sending JSON over MQTT in Python.
Start it with python manage.py mqtt_listen.
Publishing from views and tasks
Publishing is a different problem: it is short, occasional and triggered by requests. For low rates (commands, notifications) the simplest correct option is paho.mqtt.publish.single, which connects, publishes, waits for the QoS handshake and disconnects:
import json
from django.conf import settings
from django.http import JsonResponse
from django.views.decorators.http import require_POST
from paho.mqtt import publish
@require_POST
def send_command(request, device):
command = json.loads(request.body)
publish.single(
f"testmqtt/django/{device}/commands",
json.dumps(command),
qos=1,
hostname=settings.MQTT_HOST,
port=settings.MQTT_PORT,
auth={"username": settings.MQTT_USER, "password": settings.MQTT_PASS},
tls={}, # TLS with the system CA bundle
)
return JsonResponse({"queued": True})publish.single opens a new connection each time, which costs a TCP + TLS handshake per call. That is fine for a few messages per minute. If you publish many messages per second, keep one client per process instead: create it lazily, call loop_start() once, and give it a per-process client ID. The same publish.single call also works inside a Celery task, which is a good place for publishing that should not block the HTTP response.
Pushing messages to browsers with Channels
To show telemetry live, let the worker forward each message to a Channels group; every WebSocket consumer in that group receives it. Configure Channels with a Redis layer, because the worker and the ASGI server are different processes:
INSTALLED_APPS = ["daphne", *INSTALLED_APPS] # daphne first, so runserver is ASGI
ASGI_APPLICATION = "mysite.asgi.application"
CHANNEL_LAYERS = {
"default": {
"BACKEND": "channels_redis.core.RedisChannelLayer",
"CONFIG": {"hosts": [("127.0.0.1", 6379)]},
}
}The consumer joins the group and forwards events:
from channels.generic.websocket import AsyncJsonWebsocketConsumer
class TelemetryConsumer(AsyncJsonWebsocketConsumer):
async def connect(self):
await self.channel_layer.group_add("telemetry", self.channel_name)
await self.accept()
async def disconnect(self, code):
await self.channel_layer.group_discard("telemetry", self.channel_name)
async def telemetry_message(self, event): # handles type "telemetry.message"
await self.send_json(event["data"])import os
from django.core.asgi import get_asgi_application
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "mysite.settings")
django_asgi_app = get_asgi_application()
from channels.routing import ProtocolTypeRouter, URLRouter # noqa: E402
from django.urls import path # noqa: E402
from telemetry.consumers import TelemetryConsumer # noqa: E402
application = ProtocolTypeRouter({
"http": django_asgi_app,
"websocket": URLRouter([path("ws/telemetry/", TelemetryConsumer.as_asgi())]),
})Finally, extend on_message in the worker. The worker is synchronous, so wrap the async group_send with async_to_sync:
from asgiref.sync import async_to_sync
from channels.layers import get_channel_layer
channel_layer = get_channel_layer()
# inside on_message, after Reading.objects.create(...)
async_to_sync(channel_layer.group_send)(
"telemetry",
{"type": "telemetry.message", "data": {"device": device, **data}},
)In the browser, new WebSocket("ws://localhost:8000/ws/telemetry/") now receives each reading as JSON. If you add authentication, wrap the router in AuthMiddlewareStack and check self.scope["user"] in connect().
Running it in production
Treat the worker like any other daemon: one service for the ASGI server, one for mqtt_listen.
[Unit]
Description=Django MQTT worker
After=network-online.target
[Service]
WorkingDirectory=/srv/mysite
EnvironmentFile=/srv/mysite/.env
ExecStart=/srv/mysite/.venv/bin/python manage.py mqtt_listen
Restart=always
RestartSec=5
[Install]
WantedBy=multi-user.targetIf you need more than one worker for redundancy or throughput, do not just start copies: each copy would store every reading. Subscribe with a shared subscription instead ($share/django/testmqtt/django/+/telemetry) so the broker load-balances messages across the group. Shared subscriptions explains the rules and which brokers support them.
Testing the whole path
- Start Redis,
python manage.py runserverandpython manage.py mqtt_listenin separate terminals. - Open the online MQTT client, connect to the same broker and publish
{"temperature": 21.5}totestmqtt/django/sensor-1/telemetry, or use the CLI:
mosquitto_pub -h broker.connectmq.com -p 8883 \
-u "$MQTT_USER" -P "$MQTT_PASS" \
-t testmqtt/django/sensor-1/telemetry -q 1 \
-m '{"temperature": 21.5}'- Check the database:
python manage.py shell -c "from telemetry.models import Reading; print(Reading.objects.first().payload)". - Publish an invalid payload such as
helloand confirm the worker logs a warning instead of dying. - Restart the broker connection (or briefly drop your network) and confirm the worker logs a disconnect, reconnects and keeps storing messages.
Watch the Django topics in the online client
The pattern is the same for any Django project: one supervised process owns the subscription, views publish briefly, and Channels carries messages to browsers. If you are comparing frameworks, the Flask and FastAPI versions of this tutorial follow the same rules with their own lifecycle hooks.
Frequently asked questions
Where should I start the MQTT client in a Django project?
In its own process, not in AppConfig.ready() or a view. A custom management command (python manage.py mqtt_listen) that calls loop_forever() is the simplest option: it starts once, owns the subscription and can be supervised by systemd, Docker or Kubernetes.
Can I use the Django ORM inside paho-mqtt callbacks?
Yes. With loop_forever() the callbacks run in the command's main thread, so normal synchronous ORM calls work. Call close_old_connections() regularly so a long-running worker does not keep a dead database connection.
Why does Channels not deliver messages from my MQTT worker?
Most often the project uses InMemoryChannelLayer, which only works inside one process. The MQTT worker and the ASGI server are separate processes, so they need a shared layer such as channels_redis pointing at the same Redis.
