Flask ile MQTT REST API Eğitimi: Canlı Cihaz Verilerini HTTP Üzerinden Sunun

Flask ile bir MQTT REST API kurmak, HTTP konuşan her uygulamanın MQTT öğrenmeden canlı cihaz verilerini okumasını ve komut göndermesini sağlar. Bu eğitimde cihaz topic'lerine abone olan, her cihazın son değerini bellekte tutan, bunu JSON uç noktalarıyla sunan ve bir POST rotasından komut yayınlayan bir Flask uygulaması yazacaksınız. Ayrıca Flask ve MQTT birlikteliklerinin çoğunu sessizce bozan iki dağıtım hatasını da öğreneceksiniz.

MQTT'nin Önüne Neden Bir REST API Koymalı?

MQTT cihazlar için doğru protokoldür: kalıcı bağlantılar, küçük paketler ve push tabanlı teslimat. Ancak bu verilerin tüketicilerinin çoğu cihaz değildir. Bir mobil uygulamanın backend'i, bir ERP entegrasyonu, bir raporlama betiği veya bir iş ortağı sistemi genellikle HTTP üzerinden istek ve yanıt bekler. Her birine MQTT öğretmek yerine ince bir köprü çeviri yapar:

Köprü bir "son değer önbelleği"dir (last value cache). Bir historian değildir: "değer şu anda ne?" sorusunu yanıtlar, "geçen salı neydi?" sorusunu değil. Geçmiş için akışı önbelleğin yanında bir veritabanında saklayın.

Gerekenler

• Python 3.9+
• pip install "flask>=3.0" "paho-mqtt>=2.0"
• Bir broker. Geliştirme için rastgele bir topic önekiyle broker.connectmq.com adresindeki ücretsiz public MQTT broker'ı, üretim için kimlik bilgileri olan private bir broker kullanın.

Flask MQTT Köprüsü

Bunu app.py olarak kaydedin:

import json
import os
import threading
import time
import uuid
from functools import wraps

import paho.mqtt.client as mqtt
from flask import Flask, abort, jsonify, request

BROKER = os.environ.get("MQTT_HOST", "broker.connectmq.com")
PORT = int(os.environ.get("MQTT_PORT", "1883"))
USERNAME = os.environ.get("MQTT_USERNAME")
PASSWORD = os.environ.get("MQTT_PASSWORD")
CLIENT_ID = os.environ.get("MQTT_CLIENT_ID", f"flask-bridge-{uuid.uuid4().hex[:8]}")
PREFIX = os.environ.get("TOPIC_PREFIX", "connectmq/demo/flask-3391")
API_KEY = os.environ.get("API_KEY", "change-me")
STALE_AFTER = int(os.environ.get("STALE_AFTER", "120"))  # saniye

app = Flask(__name__)
cache = {}                 # cihaz -> {"data": ..., "topic": ..., "received_at": ...}
cache_lock = threading.Lock()
mqtt_connected = threading.Event()


# ---------- MQTT ----------

def on_connect(client, userdata, flags, reason_code, properties):
    if reason_code.is_failure:
        print(f"MQTT connect failed: {reason_code}")
        return
    mqtt_connected.set()
    client.subscribe(f"{PREFIX}/+/state", qos=1)


def on_disconnect(client, userdata, flags, reason_code, properties):
    mqtt_connected.clear()


def on_message(client, userdata, msg):
    device = msg.topic.split("/")[-2]
    try:
        data = json.loads(msg.payload)
    except (json.JSONDecodeError, UnicodeDecodeError):
        data = msg.payload.decode("utf-8", errors="replace")
    with cache_lock:
        cache[device] = {"data": data, "topic": msg.topic, "received_at": time.time()}


mqttc = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=CLIENT_ID)
if USERNAME:
    mqttc.username_pw_set(USERNAME, PASSWORD)
if PORT == 8883:
    mqttc.tls_set()
mqttc.on_connect = on_connect
mqttc.on_disconnect = on_disconnect
mqttc.on_message = on_message
mqttc.connect_async(BROKER, PORT, keepalive=60)
mqttc.loop_start()


# ---------- HTTP ----------

def require_api_key(view):
    @wraps(view)
    def wrapper(*args, **kwargs):
        if request.headers.get("X-API-Key") != API_KEY:
            abort(401)
        return view(*args, **kwargs)
    return wrapper


def serialize(device, entry):
    age = time.time() - entry["received_at"]
    return {
        "device": device,
        "data": entry["data"],
        "topic": entry["topic"],
        "age_seconds": round(age, 1),
        "stale": age > STALE_AFTER,
    }


@app.get("/health")
def health():
    ok = mqtt_connected.is_set()
    return jsonify({"mqtt_connected": ok}), (200 if ok else 503)


@app.get("/api/devices")
@require_api_key
def list_devices():
    with cache_lock:
        items = [serialize(d, e) for d, e in sorted(cache.items())]
    return jsonify(items)


@app.get("/api/devices/<device>")
@require_api_key
def get_device(device):
    with cache_lock:
        entry = cache.get(device)
        if entry is None:
            abort(404)
        return jsonify(serialize(device, entry))


@app.post("/api/devices/<device>/command")
@require_api_key
def send_command(device):
    if any(ch in device for ch in "/+#"):
        abort(400)
    body = request.get_json(silent=True)
    if not isinstance(body, dict) or "action" not in body:
        return jsonify({"error": "JSON body with an 'action' field is required"}), 400
    if not mqtt_connected.is_set():
        return jsonify({"error": "MQTT broker not connected"}), 503

    body["request_id"] = uuid.uuid4().hex
    info = mqttc.publish(f"{PREFIX}/{device}/cmd", json.dumps(body), qos=1)
    try:
        info.wait_for_publish(timeout=5)
    except (ValueError, RuntimeError):
        pass
    if not info.is_published():
        return jsonify({"error": "publish not acknowledged"}), 504
    return jsonify({"published": True, "request_id": body["request_id"]}), 202


if __name__ == "__main__":
    # use_reloader=False şarttır: reloader ikinci bir MQTT client'ı başlatırdı
    app.run(host="0.0.0.0", port=5000, debug=True, use_reloader=False)

Temel Uygulama Ayrıntıları

• MQTT client'ı import sırasında başlar; connect_async ve loop_start ile kendi iş parçacığında çalışır, uygulamayı ister python app.py ile ister bir WSGI sunucusuyla başlatın. connect_async ayrıca broker kısa süreliğine erişilemez olsa bile uygulamanın açılması anlamına gelir; paho yeniden denemeye devam eder.
• Önbelleği bir kilit korur. paho kendi ağ iş parçacığından yazarken Flask istek iş parçacıklarından okur.
• POST rotası 200 değil 202 Accepted döndürür. Broker mesajı onayladı (QoS 1 PUBACK), ancak bu cihazın komutu yürüttüğü anlamına gelmez. Onaya ihtiyacınız varsa cihazın aynı request_id ile bir yanıt topic'inde sonuç yayınlamasını sağlayın veya MQTT 5 özellikleri yazısında anlatılan MQTT 5 response topic modelini kullanın.
• Cihaz ID'leri doğrulanır; böylece bir çağıran +, # veya ek topic seviyeleri enjekte edemez.

Flask ile MQTT REST API Köprüsünü Deneyin

Uygulamayı başlatın:

export API_KEY=dev-secret
python app.py

Bir cihazın yapacağı gibi retained bir durum mesajı yayınlayın. Retained message'lar burada işe yarar: köprü yeniden başladığında broker her cihazın son durumunu hemen teslim eder, böylece önbellek bir sonraki raporu beklemeden dolar.

mosquitto_pub -h broker.connectmq.com -p 1883 -i flask-dev-3391 -r -q 1 \
  -t connectmq/demo/flask-3391/valve7/state -m '{"open": true, "flow": 12.4}'

HTTP üzerinden sorgulayın ve komut gönderin:

curl -s -H "X-API-Key: dev-secret" http://localhost:5000/api/devices
curl -s -H "X-API-Key: dev-secret" http://localhost:5000/api/devices/valve7
curl -s -X POST -H "X-API-Key: dev-secret" -H "Content-Type: application/json" \
  -d '{"action": "close"}' http://localhost:5000/api/devices/valve7/command

Komutun geldiğini mosquitto_sub -h broker.connectmq.com -p 1883 -i flask-sub-3391 -v -t "connectmq/demo/flask-3391/+/cmd" ile izleyin. Public alan herkes tarafından okunabildiği için işiniz bittiğinde aynı topic'e boş bir retained payload yayınlayarak retained test mesajını temizleyin.

Dağıtım Tuzakları

Bozuk Flask ve MQTT kurulumlarının çoğu şu nedenlerden biriyle başarısız olur:

1. Debug reloader kodunuzu iki kez çalıştırır. Flask'ın reloader'ı bir alt süreç başlatır; böylece iki MQTT client'ı aynı client ID ile bağlanır ve bir döngü içinde birbirinin bağlantısını keser. use_reloader=False ayarını koruyun.
2. Birden fazla WSGI worker'ı. gunicorn -w 4 uygulamayı dört kez import eder: dört MQTT client'ı ve dört ayrı önbellek; dolayısıyla art arda gelen istekler farklı değerler döndürebilir. Bunun yerine iş parçacıklarıyla tek bir worker çalıştırın: gunicorn -w 1 --threads 8 app:app.
3. Örnekler arasında sabit client ID'ler. Bir yük dengeleyicinin arkasında yatay ölçeklendirme yapıyorsanız her örneğe benzersiz bir client ID verin. Client ID'lerin kimlik bilgilerine bağlı olduğu bir broker'da her örnek için ayrı kimlik bilgileri oluşturun.
4. Önbelleği tek doğru kaynak saymak. Çevrimdışı olan bir cihazın da önbellekte bir değeri vardır. Kodun yaptığı gibi age_seconds ve stale alanlarını sunun ki tüketiciler karar verebilsin.
5. Kimlik doğrulama olmaması. POST yapabilen her şey ekipmanı çalıştırabilir. API anahtarı başlığı asgari önlemdir; internete açık her şey için uygulamayı HTTPS ve düzgün bir kimlik doğrulama katmanının arkasına koyun.

Köprüyü Genişletmek

Temel uç noktalar çalıştıktan sonra birkaç ekleme gerçek dünyadaki taleplerin çoğunu karşılar:

• Canlı güncellemeler için Server-Sent Events. GET /api/devices uç noktasını her saniye yoklamak işe yarar, ancak önbellek her değiştiğinde bir olay üreten GET /api/stream uç noktası, HTTP client'larının MQTT konuşmadan push almasını sağlar. Flask, bir generator aracılığıyla akış yanıtlarını destekler; bağlı her client'a kendi queue.Queue nesnesini verin ve güncellemeleri on_message içinden hepsine ekleyin.
• Filtreleme ve projeksiyon. ?stale=false veya ?fields=temperature gibi sorgu parametreleri mobil client'lar için yanıtları küçük tutar.
• Geçmiş. Her mesajı on_message içinde SQLite'a veya bir zaman serisi veritabanına yazın ve GET /api/devices/<id>/history?since=... ekleyin. Son değer için bellek içi önbelleği koruyun; her istekte veritabanını sorgulamaktan çok daha hızlıdır.
• OpenAPI dokümantasyonu. API'yi başka ekipler kullanıyorsa uç noktaları bir OpenAPI dosyasında tanımlayın; böylece client üretebilir ve hangi alanları bekleyeceklerini tam olarak bilirler.

Her ekleme aynı model içinde kalır: süreç başına bir MQTT client'ı, paylaşılan durumun etrafında bir kilit ve broker üzerinde asla bloklanmayan HTTP handler'ları.

Private Bir Broker'a Geçiş

Üretim için köprüyü 8883 portunda TLS ile private bir broker'a bağlayın. ConnectMQ portalında bir broker oluşturun ve üretilen kullanıcı adını, parolayı ve client ID'yi MQTT_USERNAME, MQTT_PASSWORD ve MQTT_CLIENT_ID değişkenlerine kopyalayın. Bunun gibi bir bulut MQTT broker, Flask'ın yanında Mosquitto çalıştırma ihtiyacını ortadan kaldırır; portalın canlı topic gezgini de önbelleğin neleri içermesi gerektiğini doğrulamanıza yardımcı olur.

ConnectMQ Serverless (Free) planında QoS 1'e kadar çıkar; bu da köprünün durum ve komutlar için kullandığı seviyenin ta kendisidir. Serverless (Paid) planı QoS 2 ve shared subscription ekler; ikincisi, ileride gelen trafiği paylaşması gereken birkaç köprü örneği çalıştırırsanız kullanışlıdır.

İlgili Entegrasyonlar

Bu köprü, küçük abone servislerinden oluşan bir ailenin parçasıdır. MQTT verisini Google Sheets'e gönderme elektronik tablo raporlamasını, Python ile MQTT e-posta uyarıları ise bildirimleri ele alır. Tüketicileriniz bir tarayıcıda çalışıyorsa REST'i tamamen atlayıp onları WebSocket üzerinden MQTT yazısında anlatıldığı gibi WebSocket ile bağlayabilirsiniz. Diğer diller için MQTT client kütüphaneleri rehberine bakın. Flask'ın kendi dokümantasyonu flask.palletsprojects.com adresindedir.

ConnectMQ ile Başlayın

Flask köprünüzü dakikalar içinde TLS ile korunan private bir broker'a taşıyın. app.connectmq.com adresinden kaydolun, kredi kartı gerektirmeyen ücretsiz Serverless planını seçin, bir broker oluşturun ve kimlik bilgilerini köprünün ortamına yapıştırın. Uç noktaları test ederken cihaz durumlarını yayınlamak için tarayıcı içi MQTT client'ını kullanın.

Sık Sorulan Sorular

Neden Flask-MQTT kullanılmıyor?

Kullanabilirsiniz, ancak o da aynı reloader ve çoklu worker sorunlarıyla karşılaşan ince bir sarmalayıcıdır ve hangi paho-mqtt sürümlerini desteklediğini kontrol etmeniz gerekir. paho-mqtt 2.x'i doğrudan kullanmak, iş parçacığı yönetimi ve client yaşam döngüsü üzerindeki kontrolü sizde tutar.

REST API geçmiş verileri döndürebilir mi?

Bellek içi bir önbellekle hayır. Mesajları aynı aboneden SQLite, PostgreSQL veya bir zaman serisi deposu gibi bir veritabanında saklayın ve buradan okuyan sorgu uç noktaları ekleyin.

Bir komutun yürütüldüğünü nasıl anlarım?

202 yanıtı yalnızca broker'ın mesajı kabul ettiğini doğrular. Cihazın request_id içeren bir onay yayınlamasını sağlayın ve bunu başka bir uç nokta veya cihaz durumundaki bir durum alanı aracılığıyla sunun.

Flask yerine FastAPI kullanabilir miyim?

Evet. Yapı aynıdır: süreç başına bir MQTT client'ı, kilitli bir önbellek ve onu okuyan rotalar. FastAPI ile aiomqtt gibi asenkron bir client tercih edebilirsiniz.