-
Notifications
You must be signed in to change notification settings - Fork 1
/
Copy pathflask_apiexceptions.py
244 lines (184 loc) · 7.38 KB
/
flask_apiexceptions.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
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
# -*- coding: utf-8 -*-
"""
flask_apiexceptions
~~~~~~~~~~~~~~~~~~~
Base API exception classes.
:copyright: (c) Fictive Kin.
"""
from __future__ import absolute_import
from __future__ import print_function
__version_info__ = ('1', '1', '2')
__version__ = '.'.join(__version_info__)
__author__ = 'Joel Perras'
__copyright__ = '(c) 2018 Fictive Kin, LLC'
__all__ = ['JSONExceptionHandler', 'ApiException', 'ApiError',
'api_exception_handler']
import logging
from flask import jsonify, request
from werkzeug.exceptions import default_exceptions, HTTPException
logger = logging.getLogger('apiexceptions')
class JSONExceptionHandler(object):
"""
A Flask extension that converts default Flask exceptions to their
application/json content type equivalent.
```
from application.libs import JSONExceptionHandler
exception_handler = JSONExceptionHandler()
exception_handler.init_app(app)
```
"""
# If we don't know what HTTP code to assign an exception, by default
# we assign it a `500`. This also handles uncaught exceptions; e.g.
# if our application raises any kind of Exception subclass that we don't
# explicitly have a handler for, then we've probably got an application
# error somewhere for that particular code path.
default_status_code = 500
default_message = 'An error occurred!'
def __init__(self, app=None):
"""
Initialize the extension.
Any default configurations that do not require the application
instance should be put here.
"""
if app:
self.init_app(app)
def default_handler(self, error=None):
"""Default error handler to register with the application."""
if error is None:
message = self.default_message
else:
# If the error object contains a `message` attribute, then let's
# use that as the message for our exception.
if hasattr(error, 'message'):
message = error.message
# Werkzeug default exception types use `description` instead
# of `message`.
elif hasattr(error, 'description'):
message = error.description
else:
message = self.default_message
response = jsonify(message=message)
# If our error object contains a specific error code, then let's use
# that. If not, we will use our `default_status_code` that has been
# defined for this class. This ensures that random exceptions that
# are thrown by Python or by external libraries that we miss are
# an application error.
response.status_code = self.default_status_code
if hasattr(error, 'status_code'):
response.status_code = int(error.status_code)
elif isinstance(error, HTTPException):
response.status_code = error.code
if response.status_code >= 500:
logger.exception(error)
else:
logger.debug(error)
return response
def init_app(self, app):
"""
Initialize the extension with any application-level configuration
requirements.
This is where we register the Werkzeug `HTTPException` class along
with all the other default exception codes.
"""
self.app = app
# Register the default HTTP codes to be handled by our exception
# handler.
for code, _ in default_exceptions.items():
self.register(code)
if not hasattr(self.app, 'extensions'):
self.app.extensions = {}
self.app.extensions['apiexceptions'] = self
def register(self, code_or_exception, handler=None):
"""
Register an exception class *or* numeric code with the default
exception handler provided by this extension *or* the function provided
at `handler` in the argument.
"""
f = handler or self.default_handler
self.app.register_error_handler(code_or_exception, f=f)
@staticmethod
def handle_404(error=None): #pylint: disable=locally-disabled,unused-argument
"""The default Werkzeug 404 handler does not include a
message or description, which causes some consistency issues with our
frontends when receiving a 404."""
message = 'The resource at {} could not be found.'.format(request.path)
response = jsonify(message=message)
response.status_code = 404
return response
class ApiError(object): #pylint: disable=locally-disabled,too-few-public-methods
"""
Contains information related to an API usage error.
- code: a semantically readable error slug, e.g. `invalid-password`
- info: information regarding the source of the error. E.g., if the
error was caused by invalid submitted data, the info can contain
a list of fields that contained the bad data. E.g.,
['username', 'other_field']
- message: a human-readable description of what went wrong.
All of the above data will be serialized into JSON to be returned to the
client.
"""
def __init__(self, code=None, info=None, message=None):
self.code = code
self.info = info
self.message = message
def serialize(self):
"""
Construct response dictionary.
"""
return {'code': self.code,
'info': self.info,
'message': self.message}
class ApiException(Exception):
"""
An exception that may be raised by various API view endpoints.
Can contain one or more ApiError objects and must include a HTTP status
code.
"""
# Default status code if none is set.
status_code = 500
message = None
code = None
info = None
#pylint: disable=locally-disabled,too-many-arguments
def __init__(self, status_code=None, error=None, message=None, info=None,
code=None):
"""
Initialize the ApiException container object.
If an `error` instance is provided, it will be added as an error
contained within this wrapper. If any of `message`, `info`, or `code`
are set, a new error object is created added.
"""
super(ApiException, self).__init__()
self._errors = []
if error is not None:
self._errors.append(error)
self.status_code = status_code or self.status_code
message = message or self.message
code = code or self.code
info = info or self.info
if message or info or code:
self.add_error(ApiError(message=message, code=code, info=info))
def add_error(self, error):
"""
Append an error to the list of errors contained with this
ApiException instance.
"""
self._errors.append(error)
@property
def errors(self):
"""Getter for errors currently stored on this instance."""
return self._errors
def serialize(self):
"""
Serialize the errors contained within this ApiException object to
Python types that are easily convertible to JSON (or similar).
"""
return {'errors': [e.serialize() for e in self.errors]}
def api_exception_handler(api_exception):
"""
Jsonify and serialize ApiException-compatible objects and assign
the correct response code.
"""
response = jsonify(api_exception.serialize())
response.status_code = api_exception.status_code
return response