-
-
Notifications
You must be signed in to change notification settings - Fork 753
/
flask_app.py
177 lines (144 loc) · 6.18 KB
/
flask_app.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
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
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
"""
This module defines a FlaskApp, a Connexion application to wrap a Flask application.
"""
import datetime
import logging
import pathlib
from decimal import Decimal
from types import FunctionType # NOQA
import flask
import werkzeug.exceptions
from flask import json, signals
from ..apis.flask_api import FlaskApi
from ..exceptions import ProblemException
from ..problem import problem
from .abstract import AbstractApp
logger = logging.getLogger('connexion.app')
class FlaskApp(AbstractApp):
def __init__(self, import_name, server='flask', **kwargs):
super().__init__(import_name, FlaskApi, server=server, **kwargs)
self.extra_files = kwargs.get("extra_files", [])
def create_app(self):
app = flask.Flask(self.import_name, **self.server_args)
app.json_encoder = FlaskJSONEncoder
app.url_map.converters['float'] = NumberConverter
app.url_map.converters['int'] = IntegerConverter
return app
def get_root_path(self):
return pathlib.Path(self.app.root_path)
def set_errors_handlers(self):
for error_code in werkzeug.exceptions.default_exceptions:
self.add_error_handler(error_code, self.common_error_handler)
self.add_error_handler(ProblemException, self.common_error_handler)
def common_error_handler(self, exception):
"""
:type exception: Exception
"""
signals.got_request_exception.send(self.app, exception=exception)
if isinstance(exception, ProblemException):
response = problem(
status=exception.status, title=exception.title, detail=exception.detail,
type=exception.type, instance=exception.instance, headers=exception.headers,
ext=exception.ext)
else:
if not isinstance(exception, werkzeug.exceptions.HTTPException):
exception = werkzeug.exceptions.InternalServerError()
response = problem(title=exception.name,
detail=exception.description,
status=exception.code,
headers=exception.get_headers())
return FlaskApi.get_response(response)
def add_api(self, specification, **kwargs):
api = super().add_api(specification, **kwargs)
self.app.register_blueprint(api.blueprint)
if isinstance(specification, (str, pathlib.Path)):
self.extra_files.append(self.specification_dir / specification)
return api
def add_error_handler(self, error_code, function):
# type: (int, FunctionType) -> None
self.app.register_error_handler(error_code, function)
def run(self,
port=None,
server=None,
debug=None,
host=None,
extra_files=None,
**options): # pragma: no cover
"""
Runs the application on a local development server.
:param host: the host interface to bind on.
:type host: str
:param port: port to listen to
:type port: int
:param server: which wsgi server to use
:type server: str | None
:param debug: include debugging information
:type debug: bool
:param extra_files: additional files to be watched by the reloader.
:type extra_files: Optional[Iterable[str | pathlib.Path]]
:param options: options to be forwarded to the underlying server
"""
# this functions is not covered in unit tests because we would effectively testing the mocks
# overwrite constructor parameter
if port is not None:
self.port = port
elif self.port is None:
self.port = 5000
self.host = host or self.host or '0.0.0.0'
if server is not None:
self.server = server
if debug is not None:
self.debug = debug
if extra_files is not None:
self.extra_files.extend(extra_files)
logger.debug('Starting %s HTTP server..', self.server, extra=vars(self))
if self.server == 'flask':
self.app.run(self.host, port=self.port, debug=self.debug,
extra_files=self.extra_files, **options)
elif self.server == 'tornado':
try:
import tornado.httpserver
import tornado.ioloop
import tornado.wsgi
except ImportError:
raise Exception('tornado library not installed')
wsgi_container = tornado.wsgi.WSGIContainer(self.app)
http_server = tornado.httpserver.HTTPServer(wsgi_container, **options)
http_server.listen(self.port, address=self.host)
logger.info('Listening on %s:%s..', self.host, self.port)
tornado.ioloop.IOLoop.instance().start()
elif self.server == 'gevent':
try:
import gevent.pywsgi
except ImportError:
raise Exception('gevent library not installed')
http_server = gevent.pywsgi.WSGIServer((self.host, self.port), self.app, **options)
logger.info('Listening on %s:%s..', self.host, self.port)
http_server.serve_forever()
else:
raise Exception(f'Server {self.server} not recognized')
class FlaskJSONEncoder(json.JSONEncoder):
def default(self, o):
if isinstance(o, datetime.datetime):
if o.tzinfo:
# eg: '2015-09-25T23:14:42.588601+00:00'
return o.isoformat('T')
else:
# No timezone present - assume UTC.
# eg: '2015-09-25T23:14:42.588601Z'
return o.isoformat('T') + 'Z'
if isinstance(o, datetime.date):
return o.isoformat()
if isinstance(o, Decimal):
return float(o)
return json.JSONEncoder.default(self, o)
class NumberConverter(werkzeug.routing.BaseConverter):
""" Flask converter for OpenAPI number type """
regex = r"[+-]?[0-9]*(\.[0-9]*)?"
def to_python(self, value):
return float(value)
class IntegerConverter(werkzeug.routing.BaseConverter):
""" Flask converter for OpenAPI integer type """
regex = r"[+-]?[0-9]+"
def to_python(self, value):
return int(value)