-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathflask_wrapper.py
More file actions
80 lines (63 loc) · 1.96 KB
/
flask_wrapper.py
File metadata and controls
80 lines (63 loc) · 1.96 KB
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
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
'''
Wraps a Flask server into a class so that it can be imported and ran in a separate program.
'''
import socket
import threading
import flask
def get_ip():
'''
Gets the internal IP address of the current device. OS independent.
:return: IP
'''
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
try:
s.connect(('10.255.255.255', 1))
ip = s.getsockname()[0]
except Exception as e:
print(f'{e}!r')
ip = '127.0.0.1'
finally:
s.close()
return ip
class EndpointAction():
def __init__(self, function):
self.function = function
def __call__(self, *args):
return self.function()
class FlaskWrapper():
def __init__(self, name):
'''
:param name: Flask app name
'''
self.flask_server = flask.Flask(name)
self.server_thread = None
self.host = None
self.port = None
def run(self, host=get_ip(), port=5000):
'''
Runs the flask app in a separate thread. By defualt, the app runs on the device's
internal IP.
:param host: hostname on which to run flask server (string)
:param port: port to run flask server (int)
:return: None
'''
self.host = host
self.port = port
self.server_thread = threading.Thread(
target=self.flask_server.run,
kwargs={
'host': self.host,
'port': self.port,
}
)
self.server_thread.start()
def add_endpoint(self, endpoint, name, handler, methods):
'''
Add an endpoint to the flask app
:param endpoint: url route (ex: /example)
:param name: route name
:param handler: function to call when endpoint is hit
:param methods: list of methods (POST, GET, etc)
:return: None
'''
self.flask_server.add_url_rule(endpoint, name, EndpointAction(handler), methods=methods)