Hello,
Thank you for your suggestion. I didn't know how to use webhooks at first, but now it's a bit clearer for me.
So, I tried with webhooks. And there is a problem.
My Flask server looks like this now :
class BearerAuth(AuthBase):
def __init__(self, token):
self.token = token
def __call__(self, r):
r.headers["authorization"] = "Bearer " + self.token
return r
app = Flask(__name__)
cors = CORS(app)
webhooks = {}
def getServerIp():
hostname=socket.gethostname()
IPAddr=socket.gethostbyname(hostname)
return IPAddr
def getPort():
return "8050" # test
def initWebhooks():
url = f"https://api.airtable.com/v0/bases/{baseId}/webhooks"
httpGetRequest = requests.get(url=url, auth=BearerAuth(personalToken))
response_json = httpGetRequest.json()
for w in response_json["webhooks"]:
webhooks[w["id"]] = w
print(response_json)
keysLength = list(webhooks.keys())
if len(keysLength) == 0:
# create a webhook with a notification url to the flask server, so that when something changes on the table, a notification is sent to the server
createWebhookUrl = f"https://api.airtable.com/v0/bases/{baseId}/webhooks"
serverUrl = f"https://{ipServer}:{port}/test"
data = {
"notificationUrl": serverUrl,
"specification": {
"options": {
"filters": {
"dataTypes": [
"tableData"
],
"recordChangeScope": tableId,
"fromSources": ["client", "publicApi", "automation"],
"watchDataInFieldIds": ["fldRiInNzxRPn4iAC", "fld72WfS2WokrK8b0", "fldErLsnLfSGV4ZuZ", "fldAQw2BaIOXmcd9C",
"fldJAu7jVuzJvp5jJ", "fldbogIDstibYTyuj", "fldjzycHMlwyoAdNt", "fldXZXEWAPRXUx1k4"]
}
}
}
}
httpPostRequest = requests.post(url=createWebhookUrl, json=data, auth=BearerAuth(personalToken))
print(httpPostRequest.json()["id"])
webhooks[httpPostRequest.json()["id"]].append(httpPostRequest.json())
else:
# refresh the webhook(s)
for webhook in webhooks:
webhookId = webhook
refreshWebhookUrl = f"https://api.airtable.com/v0/bases/{baseId}/webhooks/{webhookId}/refresh"
data = {}
httpPostRequest = requests.post(url=refreshWebhookUrl, data={}, auth=BearerAuth(personalToken))
ipServer = getServerIp()
port = getPort()
initWebhooks()
@app.route("/test", methods=['POST']) # test notification delivery from webhooks
def test():
# send the response with a 200 or 204 status code with an empty body
# request the webhooks payload list to get the updates
# Note: Use the decoded macSecret here, not the Base64-encoded
# version that was returned from the webhook create API action.
print(request)
#hmac = HMAC.new(webhooks[request["webhook"]["id"]]["macSecretBase64"], digestmod=SHA256);
#hmac.update(bytes(request))
#expectedContentHmac = 'hmac-sha256=' + hmac.hexdigest()
print("Hey !")
resp = Response("", status="200")
# create a thread that will fetch webhooks payloads : the thread will initiate after the response has been returned
return resp
if __name__ == '__main__':
app.run(host=ipServer, port=int(port), ssl_context='adhoc', debug=True)
After the webhook has been created, I did some modifications onto the table to check if a notification has been sent to the Flask server but I do not see any printing on the Flask console. Moreover, it looks like any private IP adresses cannot be used as a notification URL. In the part of the code where I attempt to refresh the webhook, i have this :
{'webhooks': [{'id': 'XXXXXXXXXXXX', 'specification': {'options': {'filters': {'dataTypes': ['tableData'], 'recordChangeScope': 'XXXXXXXXXXXX', 'fromSources': ['client', 'publicApi', 'automation'], 'watchDataInFieldIds': ['fldRiInNzxRPn4iAC', 'fld72WfS2WokrK8b0', 'fldErLsnLfSGV4ZuZ', 'fldAQw2BaIOXmcd9C', 'fldJAu7jVuzJvp5jJ', 'fldbogIDstibYTyuj', 'fldjzycHMlwyoAdNt', 'fldXZXEWAPRXUx1k4']}}}, 'notificationUrl': 'https://10.110.92.250:8050/test', 'cursorForNextPayload': 9, 'lastNotificationResult': {'success': False, 'completionTimestamp': '2023-05-31T12:14:53.848Z', 'durationMs': 0.330349, 'retryNumber': 6, 'error': {'message': 'The hostname resolves to an invalid or private IP address.'}, 'willBeRetried': True}, 'areNotificationsEnabled': True, 'lastSuccessfulNotificationTime': None, 'isHookEnabled': True, 'expirationTime': '2023-06-07T12:18:14.188Z'}]}
Also if I don't use "HTTPS" in the notification URL, it blocks the webhook creation.
{'error': {'type': 'INVALID_WEBHOOK_NOTIFICATION_URL', 'message': 'notificationUrl: "http:" not allowed; use "https:".'}}
I think it comes back to the same problem I had with connecting to my server via the automation script : if I rely on a private IP address for my server, I cannot communicate with Airtable services. I have the feeling the only way to do so is by using the Python API, but maybe I missed something...
