如何在Python中接收QuantumLeap/Orion订阅触发变更时的通知？

针对Orion的实现方式 #

Orion原生支持通过**Webhook（HTTP回调）**推送订阅变更通知，你可以用Python搭建一个接收服务来捕获这些通知：

1. 搭建Python Web服务（以Flask为例）
   写一个简单的POST接口，用来接收Orion推送的变更数据：

   from flask import Flask, request, jsonify
   
   app = Flask(__name__)
   
   @app.route('/orion-webhook', methods=['POST'])
   def handle_orion_notification():
       change_data = request.get_json()
       # 这里可以添加自定义处理逻辑，比如打印、存储或触发业务流程
       print("Orion订阅变更触发：", change_data)
       return jsonify({"status": "received"}), 200
   
   if __name__ == '__main__':
       # 允许外部访问，端口可自定义
       app.run(host='0.0.0.0', port=5000)
   

2. 创建Orion订阅时指定回调地址
   在创建订阅的请求中，将notification.http.url设置为上述Python服务的可访问地址（内网/公网均可，只要Orion能连通）。示例创建命令：

   curl -X POST \
     http://<orion-ip>:1026/v2/subscriptions \
     -H 'Content-Type: application/json' \
     -d '{
       "description": "监听Room1的温度变更",
       "subject": {
         "entities": [{"id": "Room1", "type": "Room"}],
         "condition": {"attrs": ["temperature"]}
       },
       "notification": {
         "http": {"url": "http://<python-service-ip>:5000/orion-webhook"},
         "attrs": ["temperature"]
       }
     }'
   

针对QuantumLeap的实现方式 #

QuantumLeap主要负责接收Orion的数据并存储，如果你需要获取它的变更通知，通常有两种途径：

方式1：监听QuantumLeap的数据库变更

QuantumLeap默认将数据存储在PostgreSQL中，你可以通过数据库的LISTEN/NOTIFY机制，用Python监听数据写入事件：

1. 在PostgreSQL中创建触发器和通知函数
   （替换<your-table>为QuantumLeap对应的存储表，比如_room_room）

   CREATE OR REPLACE FUNCTION notify_ql_update()
   RETURNS TRIGGER AS $$
   BEGIN
     -- 将新插入的数据转为JSON并发送通知
     PERFORM pg_notify('ql_data_update', row_to_json(NEW)::text);
     RETURN NEW;
   END;
   $$ LANGUAGE plpgsql;
   
   CREATE TRIGGER trigger_ql_data_insert
   AFTER INSERT ON <your-table>
   FOR EACH ROW EXECUTE FUNCTION notify_ql_update();
   

2. 用Python监听数据库通知
   使用psycopg2库连接数据库并监听指定的通知频道：

   import psycopg2
   from psycopg2.extensions import ISOLATION_LEVEL_AUTOCOMMIT
   
   # 替换为你的数据库连接信息
   db_conn = psycopg2.connect(
       dbname="quantumleap",
       user="admin",
       password="password",
       host="<db-ip>"
   )
   db_conn.set_isolation_level(ISOLATION_LEVEL_AUTOCOMMIT)
   cursor = db_conn.cursor()
   cursor.execute("LISTEN ql_data_update;")
   
   print("开始监听QuantumLeap数据变更...")
   while True:
       db_conn.poll()
       while db_conn.notifies:
           notify_msg = db_conn.notifies.pop(0)
           print("QuantumLeap数据更新：", notify_msg.payload)
   

方式2：复用Orion的Webhook

如果QuantumLeap的数据来自Orion的订阅，你可以直接在Orion的订阅中同时通知你的Python服务和QuantumLeap，这样就能在Python中直接获取原始变更通知。

内容的提问来源于stack exchange，提问作者drypatrick