FastAPI ile MQTT: HTTP ve MQTT arasında köprü

Bir FastAPI–MQTT köprüsü, sıradan HTTP istemcilerinin cihazlarla konuşmasını sağlar: bir POST MQTT publish'ine dönüşür, cihazlardan gelen mesajlar da WebSocket üzerinden tarayıcılara akar. FastAPI asyncio üzerine kurulu olduğundan bunu yapmanın en temiz yolu, uygulamanın lifespan'ında başlayan tek bir arka plan task'inin yönettiği asyncio'ya özgü bir istemcidir (aiomqtt).

Bu eğitimde küçük ama üretim düzeninde bir servis kuruyoruz: komut gönderen bir REST endpoint'i, broker yeniden başlatmalarına dayanan bir subscriber, canlı telemetriyi tarayıcılara gönderen bir WebSocket ve birden fazla worker için ölçekleme kuralları.

Mimari

Süreç başına bir MQTT bağlantısı: başlangıçta açılır ve tüm isteklerce paylaşılır. HTTP handler'ları asla istemci oluşturmaz — istek başına bir MQTT oturumu açmak yavaştır, broker'ı bağlantı talepleriyle boğar ve iki istek aynı client ID'yi kullandığı anda bozulur.

pip install "fastapi[standard]" aiomqtt

Yeniden bağlanan MQTT köprüsü

Köprü istemcinin sahibidir, bağlı olup olmadığını takip eder ve gelen mesajları bağlantı başına küçük kuyruklar aracılığıyla açık her WebSocket'e dağıtır:

import asyncio
import os
import uuid
import aiomqtt

BROKER = os.getenv("MQTT_HOST", "broker.connectmq.com")
TELEMETRY = "testmqtt/fastapi/+/telemetry"


class Hub:
    """MQTT mesajlarını WebSocket istemcilerine dağıtır."""

    def __init__(self):
        self.queues: set[asyncio.Queue] = set()

    def join(self) -> asyncio.Queue:
        q: asyncio.Queue = asyncio.Queue(maxsize=100)
        self.queues.add(q)
        return q

    def leave(self, q: asyncio.Queue):
        self.queues.discard(q)

    def broadcast(self, item: dict):
        for q in list(self.queues):
            try:
                q.put_nowait(item)
            except asyncio.QueueFull:
                pass  # yavaş tarayıcı: herkesi bloklamak yerine mesajı at


class MqttBridge:
    def __init__(self, hub: Hub):
        self.hub = hub
        self.connected = asyncio.Event()
        self.client = aiomqtt.Client(
            BROKER,
            port=1883,
            identifier=f"testmqtt-fastapi-{uuid.uuid4().hex[:8]}",  # süreç başına benzersiz
            username=os.getenv("MQTT_USER"),
            password=os.getenv("MQTT_PASS"),
        )

    async def run(self):
        delay = 1
        while True:
            try:
                async with self.client:
                    self.connected.set()
                    delay = 1
                    await self.client.subscribe(TELEMETRY, qos=1)
                    async for msg in self.client.messages:
                        self.hub.broadcast({
                            "topic": msg.topic.value,
                            "payload": msg.payload.decode(errors="replace"),
                        })
            except aiomqtt.MqttError as err:
                self.connected.clear()
                print(f"MQTT bağlantısı koptu: {err}; {delay} sn sonra yeniden denenecek")
                await asyncio.sleep(delay)
                delay = min(delay * 2, 30)

    async def publish(self, topic: str, payload: str, qos: int = 1):
        if not self.connected.is_set():
            raise ConnectionError("MQTT broker not connected")
        await self.client.publish(topic, payload, qos=qos)

run() döngüsü, aiomqtt rehberindeki yeniden bağlanma kalıbıdır: async with içinde abone ol, üstel backoff uygula ve HTTP handler'ları askıda kalmak yerine hızla hata versin diye connected bayrağını doğru tut.

FastAPI uygulaması ve lifespan

import asyncio
import contextlib
import json
from contextlib import asynccontextmanager
from typing import Annotated

from fastapi import FastAPI, HTTPException, Path, WebSocket, WebSocketDisconnect
from pydantic import BaseModel

from bridge import Hub, MqttBridge

hub = Hub()
bridge = MqttBridge(hub)


@asynccontextmanager
async def lifespan(app: FastAPI):
    task = asyncio.create_task(bridge.run())
    with contextlib.suppress(TimeoutError):
        await asyncio.wait_for(bridge.connected.wait(), timeout=10)
    yield
    task.cancel()
    with contextlib.suppress(asyncio.CancelledError):
        await task


app = FastAPI(lifespan=lifespan)
DeviceId = Annotated[str, Path(pattern=r"^[A-Za-z0-9_-]{1,64}$")]


class Command(BaseModel):
    action: str
    value: float | None = None


@app.post("/devices/{device_id}/commands", status_code=202)
async def send_command(device_id: DeviceId, cmd: Command):
    try:
        await bridge.publish(f"testmqtt/fastapi/{device_id}/commands", cmd.model_dump_json())
    except ConnectionError:
        raise HTTPException(503, "MQTT broker not connected")
    return {"queued": True, "topic": f"testmqtt/fastapi/{device_id}/commands"}


@app.get("/health")
async def health():
    return {"mqtt_connected": bridge.connected.is_set()}


@app.websocket("/ws")
async def live(websocket: WebSocket):
    await websocket.accept()
    queue = hub.join()
    try:
        while True:
            await websocket.send_text(json.dumps(await queue.get()))
    except WebSocketDisconnect:
        pass
    finally:
        hub.leave(queue)

Bilinçli birkaç tercih:

• Path parametresi doğrulanır. device_id bir topic'in parçası olur; bu yüzden bir regex, geçersiz ya da gereğinden geniş bir topic üretecek +, # ve diğer karakterleri reddeder. Doğrulanmamış kullanıcı girdisini asla topic adlarına yerleştirmeyin.
• 200 değil 202 Accepted. Başarılı bir QoS 1 publish, mesajın broker'da olduğu anlamına gelir — cihazın komutu çalıştırdığı anlamına gelmez. Onay gerekiyorsa cihazın bir response topic'e yanıt vermesini sağlayın.
• Başlangıç ilk bağlantı için en fazla 10 saniye bekler, ama broker kapalıysa uygulamayı düşürmez; /health durumu bildirir, kararı orkestratörünüz verir.
• Sınırlı kuyruklar sayesinde WebSocket başına yavaş bir tarayıcı, diğer herkese teslimatı yavaşlatamaz.

fastapi dev main.py (veya uvicorn main:app) ile çalıştırın.

Tarayıcıya push

Tarayıcının yalnızca standart WebSocket API'sine ihtiyacı vardır — doğrudan MQTT konuşmaz, dolayısıyla broker kimlik bilgilerini frontend'e açmazsınız:

<ul id="log"></ul>
<script>
  const ws = new WebSocket(`ws://${location.host}/ws`);
  ws.onmessage = (event) => {
    const { topic, payload } = JSON.parse(event.data);
    const li = document.createElement("li");
    li.textContent = `${topic}: ${payload}`;
    document.getElementById("log").prepend(li);
  };
</script>

Tarayıcının broker'a kendisinin abone olmasını tercih ederseniz bunun yerine 8084 portunda WebSocket üzerinden MQTT kullanın — bkz. WebSocket üzerinden MQTT. Köprü yaklaşımı, API'nin mesajları kullanıcılar görmeden önce filtrelemesini, yetkilendirmesini veya zenginleştirmesini istediğinizde daha iyidir; bunun üzerine kurulu tam bir arayüz için canlı dashboard eğitimine bakın.

Köprüyü uçtan uca test etmek

1. Uygulamayı başlatın ve curl localhost:8000/health komutunun {"mqtt_connected": true} döndürdüğünü kontrol edin.
2. Online MQTT istemcisini açın, broker.connectmq.com'a bağlanın ve testmqtt/fastapi/# topic'ine abone olun.
3. HTTP üzerinden bir komut gönderin ve online istemcide göründüğünü izleyin:

curl -X POST localhost:8000/devices/sensor-1/commands \
  -H 'content-type: application/json' \
  -d '{"action": "set_interval", "value": 30}'

1. WebSocket'e bağlanın (örneğin websocat ws://localhost:8000/ws ile), ardından online istemciden testmqtt/fastapi/sensor-1/telemetry topic'ine {"temp": 22.4} publish edin. websocat bunu hemen yazdırmalıdır.
2. Broker'ınızı yeniden başlatın ya da ağı kesin: /health false olmalı, komut endpoint'i 503 döndürmeli ve ikisi de kendiliğinden toparlanmalıdır.

FastAPI köprüsünün trafiğini izleyin

Otomatik testlerde with TestClient(app) as client: lifespan'ı çalıştırır; köprü üretimdeki gibi bağlanır. MQTT_HOST'u bir container içindeki broker'a yönlendirin. Hazır fixture'lar MQTT pytest fixture'ları yazısında.

Birden fazla worker çalıştırmak

uvicorn --workers 4 veya Gunicorn birden fazla süreç başlatır ve her biri lifespan'ı çalıştırır — yani dört MQTT bağlantısı elde edersiniz. Bunu planlarsanız sorun değildir:

• Süreç başına benzersiz client ID. Köprü rastgele bir sonek ekler. Sabit bir ID ile worker'lar bir döngü içinde birbirini düşürürdü; bkz. yinelenen client ID.
• WebSocket fan-out'u, her worker'ın her mesajı almasını gerektirir, çünkü bir tarayıcı herhangi birine bağlı olabilir. Normal abonelikler tam olarak bunu yapar.
• İşleme yalnızca bir kez olmalı. Bir worker telemetriyi aynı zamanda veritabanına yazıyorsa, o tüketici için $share/api/testmqtt/fastapi/+/telemetry gibi bir shared subscription kullanın; böylece broker mesajları çoğaltmak yerine worker'lar arasında dağıtır. Bkz. shared subscription'lar.

Alternatifler

• loop_start() ile paho-mqtt çalışır, ancak callback'leri ayrı bir thread'de çalışır. Her mesajı loop.call_soon_threadsafe(...) ile aktarın ve callback içinden asyncio nesnelerine asla doğrudan dokunmayın.
• fastapi-mqtt, gmqtt üzerine kurulu ve abonelikler için decorator'lar ekleyen bir topluluk eklentisidir. Tekrarlayan kodu azaltır; yukarıdaki açık köprü ise yeniden bağlanma ve geri basınç (back-pressure) üzerinde daha fazla kontrol verir.

Bir FastAPI–MQTT köprüsü özünde şudur: süreç başına lifespan içinde yönetilen uzun ömürlü tek bir bağlantı, dürüst bir sağlık durumu ve tarayıcılara giden sınırlı kuyruklar. Topic'e dönüşen her şeyi doğrulayın, client ID'leri worker başına benzersiz tutun ve cihazları arkasına bağlamadan önce tüm döngüyü — HTTP giriş, MQTT çıkış, MQTT giriş, WebSocket çıkış — test edin.

Sıkça sorulan sorular