MQTT avec Django : worker en management command et Channels

La façon fiable d'utiliser MQTT avec Django est de garder la connexion MQTT hors du processus web. Exécutez un abonné de longue durée sous forme de management command personnalisée, laissez-le écrire en base avec l'ORM habituel, publiez depuis les vues avec un client éphémère et utilisez Django Channels lorsque les navigateurs ont besoin des messages en direct.

Ce tutoriel construit exactement cela avec paho-mqtt 2.x, puis montre comment déployer le worker et tester tout le parcours de bout en bout.

Pourquoi ne pas se connecter dans apps.py ou une vue ?

Le premier réflexe est souvent de démarrer un client paho dans AppConfig.ready(). Cela marche sur votre portable, puis casse de manière subtile :

Où vous démarrez le client	Ce qui se passe mal
`AppConfig.ready()`	S'exécute dans chaque processus qui charge Django : `runserver` (deux fois, à cause de l'autoreloader), chaque worker Gunicorn, `migrate`, `shell`, les workers Celery. Chacun ouvre une connexion.
Dans une vue	Une nouvelle session MQTT par requête HTTP : lent, inonde le broker de CONNECT et échoue dès que deux requêtes partagent un client ID.
Une management command	Exactement un processus que vous lancez volontairement, facile à superviser et à redémarrer.

Plusieurs processus abonnés au même topic signifie que chaque message est traité plusieurs fois, et s'ils partagent un client ID, le broker les déconnecte en boucle (voir client ID dupliqué). Donc : abonnez-vous dans un worker, publiez là où vous en avez besoin.

Mise en place du projet

Installation

pip install django paho-mqtt channels channels-redis daphne
django-admin startproject mysite .
python manage.py startapp telemetry

Ajoutez telemetry à INSTALLED_APPS ainsi qu'un modèle pour les mesures reçues :

telemetry/models.py

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"]

Gardez les paramètres de connexion dans settings.py, lus depuis l'environnement :

mysite/settings.py (extrait)

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"

Lancez python manage.py makemigrations telemetry && python manage.py migrate.

Le worker MQTT en management command

Créez telemetry/management/commands/mqtt_listen.py (management/ et commands/ ont chacun besoin d'un __init__.py vide) :

telemetry/management/commands/mqtt_listen.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()  # CA système, vérifie le nom d'hôte
        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()  # bloquant, reconnexion automatique
        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
        # S'abonner ici restaure l'abonnement après chaque reconnexion
        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()  # ferme les connexions BD plus vieilles que 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)

Quelques détails comptent ici :

Abonnez-vous dans on_connect, pas après connect(). paho se reconnecte tout seul, et s'abonner dans le callback restaure l'abonnement à chaque fois.
loop_forever() exécute les callbacks dans le thread principal, donc les appels ORM synchrones sont sûrs. Pas besoin de sync_to_async dans ce processus.
close_old_connections() est ce que Django appelle au début et à la fin de chaque requête. Un worker ne voit jamais de requête : appelez-le vous-même, sinon le processus finira par échouer sur une connexion fermée après un redémarrage de la base.
Un client ID unique par processus (nom d'hôte + PID). Pour un autre schéma, utilisez le générateur de client ID.
Ne plantez jamais sur un payload invalide. Une exception non gérée dans un callback arrête la boucle. Pour l'analyse et la validation, voir envoyer du JSON via MQTT en Python.

Lancez-le avec python manage.py mqtt_listen.

Publier depuis les vues et les tâches

Publier est un autre problème : c'est bref, occasionnel et déclenché par des requêtes. Pour de faibles débits (commandes, notifications), l'option correcte la plus simple est paho.mqtt.publish.single, qui se connecte, publie, attend l'échange QoS et se déconnecte :

telemetry/views.py

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 avec le magasin de CA du système
    )
    return JsonResponse({"queued": True})

publish.single ouvre une nouvelle connexion à chaque appel, soit un handshake TCP + TLS par message. C'est acceptable pour quelques messages par minute. Si vous publiez beaucoup de messages par seconde, gardez plutôt un client par processus : créez-le à la demande, appelez loop_start() une fois et donnez-lui un client ID propre au processus. Le même appel publish.single fonctionne aussi dans une tâche Celery, un bon endroit pour les publications qui ne doivent pas bloquer la réponse HTTP.

Pousser les messages vers le navigateur avec Channels

Pour afficher la télémétrie en direct, le worker transmet chaque message à un groupe Channels ; chaque consumer WebSocket du groupe le reçoit. Configurez Channels avec une couche Redis, car le worker et le serveur ASGI sont des processus différents :

mysite/settings.py (extrait)

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)]},
    }
}

Le consumer rejoint le groupe et transmet les événements :

telemetry/consumers.py

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"])

mysite/asgi.py

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())]),
})

Enfin, étendez on_message dans le worker. Le worker est synchrone : enveloppez le group_send asynchrone avec async_to_sync :

mqtt_listen.py (on_message étendu)

from asgiref.sync import async_to_sync
from channels.layers import get_channel_layer

channel_layer = get_channel_layer()

# dans on_message, après Reading.objects.create(...)
async_to_sync(channel_layer.group_send)(
    "telemetry",
    {"type": "telemetry.message", "data": {"device": device, **data}},
)

Dans le navigateur, new WebSocket("ws://localhost:8000/ws/telemetry/") reçoit désormais chaque mesure en JSON. Si vous ajoutez de l'authentification, enveloppez le routeur dans AuthMiddlewareStack et vérifiez self.scope["user"] dans connect().

Mise en production

Traitez le worker comme n'importe quel daemon : un service pour le serveur ASGI, un autre pour mqtt_listen.

/etc/systemd/system/mqtt-listen.service

[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.target

Si vous avez besoin de plusieurs workers pour la redondance ou le débit, ne lancez pas simplement des copies : chacune enregistrerait toutes les mesures. Abonnez-vous plutôt avec une shared subscription ($share/django/testmqtt/django/+/telemetry) pour que le broker répartisse les messages dans le groupe. Shared subscriptions explique les règles et quels brokers les prennent en charge.

Tester tout le parcours

Démarrez Redis, python manage.py runserver et python manage.py mqtt_listen dans des terminaux séparés.
Ouvrez le client MQTT en ligne, connectez-vous au même broker et publiez {"temperature": 21.5} sur testmqtt/django/sensor-1/telemetry, ou utilisez la CLI :

Publier une mesure de test

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}'

Vérifiez la base : python manage.py shell -c "from telemetry.models import Reading; print(Reading.objects.first().payload)".
Publiez un payload invalide comme hello et vérifiez que le worker journalise un avertissement au lieu de s'arrêter.
Coupez la connexion au broker (ou brièvement votre réseau) et vérifiez que le worker journalise la déconnexion, se reconnecte et continue d'enregistrer les messages.

Observer les topics Django dans le client en ligne

Le schéma est le même pour tout projet Django : un processus supervisé possède l'abonnement, les vues publient brièvement et Channels achemine les messages vers les navigateurs. Si vous comparez les frameworks, les versions Flask et FastAPI de ce tutoriel suivent les mêmes règles avec leurs propres hooks de cycle de vie.

Questions fréquentes

Où démarrer le client MQTT dans un projet Django ?

Dans son propre processus, pas dans AppConfig.ready() ni dans une vue. Une management command personnalisée (python manage.py mqtt_listen) qui appelle loop_forever() est l'option la plus simple : elle démarre une seule fois, possède l'abonnement et peut être supervisée par systemd, Docker ou Kubernetes.

Puis-je utiliser l'ORM Django dans les callbacks paho-mqtt ?

Oui. Avec loop_forever(), les callbacks s'exécutent dans le thread principal de la commande, donc les appels ORM synchrones fonctionnent. Appelez régulièrement close_old_connections() pour qu'un worker de longue durée ne garde pas une connexion à la base de données morte.

Pourquoi Channels ne délivre-t-il pas les messages de mon worker MQTT ?

Le plus souvent, le projet utilise InMemoryChannelLayer, qui ne fonctionne qu'à l'intérieur d'un seul processus. Le worker MQTT et le serveur ASGI sont des processus distincts : ils ont besoin d'une couche partagée comme channels_redis pointant vers le même Redis.

MQTT avec Django : workers en arrière-plan et Channels