blob: 0faf9da01b49c3f3f20c818fc18c023865b82695 (
plain)
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
|
import threading
from typing import Any, Callable
from wsgiref.simple_server import make_server
import port_for
from flask import Flask, abort, request
class TriggerServerThread(threading.Thread):
"""Simple Flask server (wrapped as thread) for triggering actions on sessions."""
def __init__(self, trigger_method: Callable[[str], Any]):
super().__init__()
self.port = port_for.select_random()
app = Flask(__name__)
@app.route("/", methods=["POST"])
def trigger():
json = request.json
token = json.get("token", "")
r = trigger_method(token)
if r:
return "Successful triggered"
else:
return abort(403)
self.srv = make_server("0.0.0.0", self.port, app)
self.ctx = app.app_context()
self.ctx.push()
def run(self):
self.srv.serve_forever()
def shutdown(self):
self.srv.shutdown()
|
