-
Notifications
You must be signed in to change notification settings - Fork 5
/
main.py
43 lines (32 loc) · 1.2 KB
/
main.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
#py 3.7
import cherrypy
from cherrypy.process import servers
from fetch import send_to_compute
#Cherrypy config
def fake_wait_for_occupied_port(host, port): return
servers.wait_for_occupied_port = fake_wait_for_occupied_port
#If we want to implement webhook authentication
secret = "compute"
#Setup the app
class WebhookServer():
def __init__(self) -> None:
super().__init__()
@cherrypy.expose
@cherrypy.tools.json_in()
# This function will be called if the webhook endpoint is accessed
def webhook(self, *args, **kwargs):
payload_json = cherrypy.request.json.get('payload', '{}')
signature = cherrypy.request.headers["X-WEBHOOK-SIGNATURE"]
print("Webhook Signal Received")
self.webhook_called(payload_json, signature)
# This function will trigger the compute server
def webhook_called(self, payload_json: str, signature: str):
print("Compute Called")
compute_sender = send_to_compute()
print("Compute Answered")
cherrypy.config.update({
#'environment': 'production',
'server.socket_host': '0.0.0.0',
'server.socket_port': 8080
})
cherrypy.quickstart(WebhookServer())