Newer
Older
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
import logging, time
from typing import Any, Optional
from flask import Flask, request
from flask_restful import Api, Resource
from flask_socketio import Namespace, SocketIO
LOGGER = logging.getLogger(__name__)
def log_request(response):
timestamp = time.strftime('[%Y-%b-%d %H:%M]')
LOGGER.info(
'%s %s %s %s %s', timestamp, request.remote_addr, request.method,
request.full_path, response.status
)
return response
class NbiApplication:
def __init__(self, base_url : Optional[str] = None) -> None:
if base_url is None: base_url = ''
self.base_url = base_url
self.app = Flask(__name__)
self.app.after_request(log_request)
self.api = Api(self.app, prefix=base_url)
#websocket_path = '/'.join([base_url.rstrip('/'), 'websocket'])
self.sio = SocketIO(self.app, path=base_url, cors_allowed_origins="*")
def add_rest_api_resource(self, resource_class : Resource, *urls, **kwargs) -> None:
self.api.add_resource(resource_class, *urls, **kwargs)
def add_websocket_namespace(self, namespace_class : Namespace, namespace_url : str) -> None:
self.sio.on_namespace(namespace_class(namespace=namespace_url))
def websocket_emit_message(
self, event : str, *args : Any, namespace : str = "/", to : Optional[str] = None
) -> None:
self.sio.emit(event, *args, namespace=namespace, to=to)
def dump_configuration(self) -> None:
LOGGER.debug('Configured Resources:')
for resource in self.api.resources:
LOGGER.debug(' - {:s}'.format(str(resource)))
LOGGER.debug('Configured Rules:')
for rule in self.app.url_map.iter_rules():
LOGGER.debug(' - {:s}'.format(str(rule)))
def run_standalone(
self, bind_address : Optional[str] = None, bind_port : Optional[int] = None
) -> None:
# Run method used when started in a standalone mode, i.e., outside gunicorn or
# similar WSGI HTTP servers. Otherwise, use mechanism defined by the used
# WSGI HTTP server.
#logging.getLogger('werkzeug').setLevel(logging.WARNING)
endpoint = 'http://{:s}:{:s}'.format(str(bind_address), str(bind_port))
if self.base_url is not None:
endpoint = '/'.join([endpoint.rstrip('/'), self.base_url])
LOGGER.info('Listening on {:s}...'.format(endpoint))
self.sio.run(self.app, host=bind_address, port=bind_port)