如何创建 python webhook 发件人应用程序？

Question

这是post 的后续问题。

我有一个通过xxx.com\data API 端点公开的数据仓库表

我一直在使用以下代码查询此表并将其解析为数据框，如下所示；

import requests
import json
import http.client
import pandas as pd

url = "xxx.com\data?q=Active%20%3D1%20and%20LATITUDE%20%3D%20%20%220.000000%22%20and%20LONGITUDE%20%3D%20%220.000000%22&pageSize =300"
payload = {}
headers = {'Authorization': access_token}
response = requests.request("GET", url, headers=headers, data = payload)
j=json.loads(response.text.encode('utf8'))
df = pd.json_normalize(j['DataSet'])

仓库表会定期更新，我需要创建一个 webhook 以供以下 Azure httptrigger 监听；

import logging
import os
import json
import pandas as pd
import azure.functions as func

def main(req: func.HttpRequest) -> func.HttpResponse:
    logging.info('Python HTTP trigger function processed a request.')
    d={
    'Date' :['2016-10-30','2016-10-30','2016-11-01','2016-10-30'],
    'Time':['09:58:11', '10:05:34', '10:07:57', '11:15:32'],
    'Transaction':[2,3,1,1]
    }
    df=pd.DataFrame(d, columns=['Date','Time','Transaction'])
    output = df.to_csv (index_label="idx", encoding = "utf-8")
 
return func.HttpResponse(output)

运行时，httptrigger 成功侦听了我创建并在我的磁盘上本地运行的以下 webhooker 发送器。

    import logging
    import os
    import json
    import pandas as pd

data={'Lat': '0.000000',
   'Long': '0.000000',
   'Status': '1', 'Channel URL':"xxx.com\data"}

webhook_url="http://localhost:7071/api/HttpTrigger1"

r=requests.post(webhook_url, headers={'Content-Type':'application/json'}, data =json.dumps(l))

我的问题是；

1. 如何将 webhook 发送器作为应用程序部署到云中，以便每次使用 Lat==0、Long=00 和 Status=1 更新“xxx.com\data”时，都会向我的 webhook 侦听器发送一条消息？

该应用可以是 Azure/Flask/postman 或任何其他基于 python 的 webhook 构建器。

Answer 1

一种简单的方法是将您的发件人代码包装到 Timer Trigger Function 中，这将每隔 x 秒（或您决定的任何频率）轮询您的 xxx.com\data 并调用您的 webhook（另一个 http 触发函数）（如果有）改变。

{
    "name": "mytimer",
    "type": "timerTrigger",
    "direction": "in",
    "schedule": "0 */5 * * * *"
}

import datetime
import logging
import os
import json
import pandas as pd

import azure.functions as func


def main(mytimer: func.TimerRequest) -> None:
    utc_timestamp = datetime.datetime.utcnow().replace(
        tzinfo=datetime.timezone.utc).isoformat()

    if mytimer.past_due:
        logging.warn('The timer is past due!')
 
    # "xxx.com\data" polling in real scenario
    l={'Lat': '0.000000',
          'Long': '0.000000',
          'Status': '1', 
          'Channel URL':"xxx.com\data"}

    webhook_url="{function app base url}/api/HttpTrigger1"

    r=requests.post(webhook_url, headers={'Content-Type':'application/json'}, data =json.dumps(l))

    logging.info('Python timer trigger function ran at %s', utc_timestamp)

最终，您可以将 webhook 函数（http 触发器）和发送者（定时器触发器轮询）部署到 Function 应用程序中。

您还可以考虑完全摆脱 webhook 函数（以节省一个中间跃点），并将您的工作放入同一个计时器触发函数中。

Answer 2

1. 您目前有一些轮询逻辑（在querying this table using the following code 下）。如果您想将其移至“Cloud”，请使用 create a TimerTrigger 函数，并将所有轮询代码放入其中。

2. 如果您想保持该轮询器代码不变，但希望在轮询器检测到更改时调用“云”中的某些代码（更新为Lat==0,Long=00 and Status=1），那么您可以创建一个HTTPTrigger 函数并从轮询器调用它每当它检测到变化时。

---

令人困惑的部分是：您今天如何检测到这种变化？ poller 代码托管在哪里，执行频率如何？

如果数据库中的数据发生变化，那么只要数据发生变化，您可以执行“一些代码”的唯一方法是：

1. 定期轮询数据库，比如每 1 分钟一次，如果有变化，请执行“一些代码”或
2. 此数据库的某些功能允许您配置一个 REST API (HTTP Webhook)，只要有更改，该数据库就会调用该 API。实现一个 REST API（例如，作为 HttpTrigger 函数）并将您想要执行的“一些代码”放入其中。现在，只要有更改，数据库就会调用您的 webhook/REST-API 并执行“一些代码”。

读取它的方法是调用 REST API (xxx.com/data?q=...)，这是您可以检测到的唯一方法