def make_json_app(import_name, **kwargs):
"""
Creates a JSON-oriented Flask app.
All error responses that you don't specifically
manage yourself will have application/json content
type, and will contain JSON like this (just an example):
{ "message": "405: Method Not Allowed" }
"""
def make_json_error(ex):
response = jsonify(message=str(ex))
response.status_code = (ex.code if isinstance(ex, HTTPException) else 500)
return response
app = Flask(import_name, **kwargs)
for code in default_exceptions.iterkeys():
app.register_error_handler(code, make_json_error)
return app
python类HTTPException()的实例源码
def handle_user_exception(self, e):
"""
override handle_user_exception and redirect the exception to handle_api_exception
"""
exc_type, exc_value, tb = sys.exc_info()
assert exc_value is e
if isinstance(e, APIError):
return self.handle_api_exception(e)
# hook HttpException and return handle_api_exception
if isinstance(e, HTTPException) and not self.trap_http_exception(e):
# return self.handle_http_exception(e)
return self.handle_api_exception(e)
handler = self._find_error_handler(e)
if handler is None:
reraise(exc_type, exc_value, tb)
return handler(e)
def test(self, path_info=None, method=None):
"""Test if a rule would match. Works like `match` but returns `True`
if the URL matches, or `False` if it does not exist.
:param path_info: the path info to use for matching. Overrides the
path info specified on binding.
:param method: the HTTP method used for matching. Overrides the
method specified on binding.
"""
try:
self.match(path_info, method)
except RequestRedirect:
pass
except HTTPException:
return False
return True
def test(self, path_info=None, method=None):
"""Test if a rule would match. Works like `match` but returns `True`
if the URL matches, or `False` if it does not exist.
:param path_info: the path info to use for matching. Overrides the
path info specified on binding.
:param method: the HTTP method used for matching. Overrides the
method specified on binding.
"""
try:
self.match(path_info, method)
except RequestRedirect:
pass
except HTTPException:
return False
return True
def _process_event(self, repo, event):
"""Process potentially new event for repository
:param repo: Repository related to event
:type repo: ``repocribro.models.Repository``
:param event: GitHub event data
:type event: dict
:return: If the event was new or already registered before
:rtype: bool
"""
last = pytz.utc.localize(repo.last_event)
if iso8601.parse_date(event['created_at']) <= last:
return False
hook_type = self.event2webhook.get(event['type'], 'uknown')
for event_processor in self.hooks.get(hook_type, []):
try:
event_processor(db=self.db, repo=repo,
payload=event['payload'],
actor=event['actor'])
print('Processed {} from {} event for {}'.format(
event['type'], event['created_at'], repo.full_name
))
except HTTPException:
print('Error while processing #{}'.format(event['id']))
return True
def test(self, path_info=None, method=None):
"""Test if a rule would match. Works like `match` but returns `True`
if the URL matches, or `False` if it does not exist.
:param path_info: the path info to use for matching. Overrides the
path info specified on binding.
:param method: the HTTP method used for matching. Overrides the
method specified on binding.
"""
try:
self.match(path_info, method)
except RequestRedirect:
pass
except HTTPException:
return False
return True
def test(self, path_info=None, method=None):
"""Test if a rule would match. Works like `match` but returns `True`
if the URL matches, or `False` if it does not exist.
:param path_info: the path info to use for matching. Overrides the
path info specified on binding.
:param method: the HTTP method used for matching. Overrides the
method specified on binding.
"""
try:
self.match(path_info, method)
except RequestRedirect:
pass
except HTTPException:
return False
return True
def make_json_app(import_name, **kwargs):
app = flask.Flask(import_name, **kwargs)
@app.errorhandler(exceptions.FuxiException)
@app.errorhandler(cinder_exception.ClientException)
@app.errorhandler(nova_exception.ClientException)
@app.errorhandler(manila_exception.ClientException)
@app.errorhandler(processutils.ProcessExecutionError)
@app.errorhandler(brick_exception.BrickException)
def make_json_error(ex):
LOG.error("Unexpected error happened: %s",
traceback.format_exc())
response = flask.jsonify({"Err": str(ex)})
response.status_code = w_exceptions.InternalServerError.code
if isinstance(ex, w_exceptions.HTTPException):
response.status_code = ex.code
content_type = 'application/vnd.docker.plugins.v1+json; charset=utf-8'
response.headers['Content-Type'] = content_type
return response
for code in w_exceptions.default_exceptions:
app.register_error_handler(code, make_json_error)
return app
def can_handle_request(self, environ):
'''
Decides whether it can handle a request with the Flask app by
matching the request environ against the route mapper
Returns (True, 'flask_app') if this is the case.
'''
# TODO: identify matching urls as core or extension. This will depend
# on how we setup routing in Flask
urls = self.url_map.bind_to_environ(environ)
try:
endpoint, args = urls.match()
log.debug('Flask route match, endpoint: {0}, args: {1}'.format(
endpoint, args))
return (True, self.app_name)
except HTTPException:
return (False, self.app_name)
def test(self, path_info=None, method=None):
"""Test if a rule would match. Works like `match` but returns `True`
if the URL matches, or `False` if it does not exist.
:param path_info: the path info to use for matching. Overrides the
path info specified on binding.
:param method: the HTTP method used for matching. Overrides the
method specified on binding.
"""
try:
self.match(path_info, method)
except RequestRedirect:
pass
except HTTPException:
return False
return True
def test(self, path_info=None, method=None):
"""Test if a rule would match. Works like `match` but returns `True`
if the URL matches, or `False` if it does not exist.
:param path_info: the path info to use for matching. Overrides the
path info specified on binding.
:param method: the HTTP method used for matching. Overrides the
method specified on binding.
"""
try:
self.match(path_info, method)
except RequestRedirect:
pass
except HTTPException:
return False
return True
def test(self, path_info=None, method=None):
"""Test if a rule would match. Works like `match` but returns `True`
if the URL matches, or `False` if it does not exist.
:param path_info: the path info to use for matching. Overrides the
path info specified on binding.
:param method: the HTTP method used for matching. Overrides the
method specified on binding.
"""
try:
self.match(path_info, method)
except RequestRedirect:
pass
except HTTPException:
return False
return True
def raise_exc(cls, code, message):
"""
Raise an HTTPException with a code and a message sent in a json like
{
"code": code
"message": message
}
:param cls: HTTPException of the error, for example NotFound, BadRequest, NotAuthorized
:param code: A brief message for the exception, like MISSING_PARAMETER
:param message: A longer description of the error
:return: Nothing, raise the provided exception with the correct response
"""
response = Response()
response.mimetype = "application/json"
response.status_code = cls.code
response.data = json.dumps({
"code": code,
"message": message
})
Logger.warning(cls.__name__.upper(), code + ": " + message)
raise cls(response=response)
def handle(self, endpoint, route_args, request):
"""
Handle a request in the derived handler. The request is routed to the correct method using *endpoint*
:param endpoint: A string with the name of the class method to call with (route_args, request) as parameters,
this method should return a Response or call self.raise_exc. *NOTE*: the method MUST be implemented in the
derived class
:param route_args: The route parameters, the parameters extracted from the matching route in the URL
:param request: The Request object, request.args contains the query parameters of the request
:return: Return a Response if the request is successful, an HTTPException if an error occurred
"""
try:
data = BaseHandler._call(self.__getattribute__(endpoint), route_args, request)
response = Response()
if data is not None:
response.code = 200
response.mimetype = "application/json"
response.data = json.dumps(data)
else:
response.code = 204
return response
except HTTPException as e:
return e
def test(self, path_info=None, method=None):
"""Test if a rule would match. Works like `match` but returns `True`
if the URL matches, or `False` if it does not exist.
:param path_info: the path info to use for matching. Overrides the
path info specified on binding.
:param method: the HTTP method used for matching. Overrides the
method specified on binding.
"""
try:
self.match(path_info, method)
except RequestRedirect:
pass
except HTTPException:
return False
return True
def http_error_handler(error, without_code=False):
if isinstance(error, HTTPException):
error = error.code
elif not isinstance(error, int):
error = 500
body = render_template('errors/{}.html'.format(error))
if not without_code:
return make_response(body, error)
else:
return make_response(body)
# -----------------------------------------------------------
# Hooks
def test_SupervisorRolePermission_authenticated_user_with_password_with_check_supervisor(
authenticated_user_instance
):
authenticated_user_instance.password = "correct_password"
obj = Mock()
obj.check_supervisor = lambda user: user == authenticated_user_instance
with permissions.SupervisorRolePermission(
obj=obj,
password_required=True,
password="correct_password"
):
pass
with pytest.raises(HTTPException):
with permissions.SupervisorRolePermission(
obj=obj,
password_required=True,
password="wrong_password"
):
pass
def test_SupervisorRolePermission_authenticated_user_with_password_without_check_supervisor(
authenticated_user_instance
):
authenticated_user_instance.password = "correct_password"
obj = Mock()
del obj.check_supervisor
with pytest.raises(HTTPException):
with permissions.SupervisorRolePermission(
obj=obj,
password_required=True,
password="correct_password"
):
pass
with pytest.raises(HTTPException):
with permissions.SupervisorRolePermission(
obj=obj,
password_required=True,
password="wrong_password"
):
pass
def test_OwnerRolePermission_authenticated_user_with_password_without_check_owner(
authenticated_user_instance
):
authenticated_user_instance.password = "correct_password"
obj = Mock()
del obj.check_owner
with pytest.raises(HTTPException):
with permissions.OwnerRolePermission(
obj=obj,
password_required=True,
password="correct_password"
):
pass
with pytest.raises(HTTPException):
with permissions.OwnerRolePermission(
obj=obj,
password_required=True,
password="wrong_password"
):
pass
def _handle_error(self, e):
"""Returns handled excpetion. Detects blueprint from global
:class:`~flask.wrappers.Request` object, and passes exception object to
its `registered_errorhandler`.
"""
blueprint = self._detect_blueprint()
if isinstance(e, HTTPException):
if self._error_loader_callback is not None:
e = self._error_loader_callback(e) # load custom exception
else:
print(traceback.format_exc())
e = HTTPException()
if blueprint is not None:
if blueprint.name in self._errorhandlers:
return self._errorhandlers[blueprint.name](e)
else:
if None in self._errorhandlers:
return self._errorhandlers[None](e)
return e
def test(self, path_info=None, method=None):
"""Test if a rule would match. Works like `match` but returns `True`
if the URL matches, or `False` if it does not exist.
:param path_info: the path info to use for matching. Overrides the
path info specified on binding.
:param method: the HTTP method used for matching. Overrides the
method specified on binding.
"""
try:
self.match(path_info, method)
except RequestRedirect:
pass
except HTTPException:
return False
return True
def handle_exception(error):
code = 500
message = None
if hasattr(error, 'status_code') :
code = error.status_code
if hasattr(error, 'message') :
message = str(error.message)
if isinstance(error, HTTPException):
code = error.code
message = str(error)
extra = error.extra if hasattr(error, 'extra') else None
response = jsonify(format_exception(message, code=code, extra=extra))
response.status_code = code
return response
def error_handler(error):
"""
Standard Error Handler
"""
if isinstance(error, HTTPException):
return jsonify({
'statusCode': error.code,
'name': error.name,
'description': error.description
}), error.code
else:
return jsonify({
'statusCode': 500,
'name': 'Internal Server Error',
'description': 'An unknown error has occurred'
}), 500
# common errors - add others as needed
def render(self, exception):
"""
Render the exception
:param exception: The exception
:type exception: Exception
:return: The response
"""
if not isinstance(exception, HTTPException):
http_exception = InternalServerError()
else:
http_exception = exception
is_server_error = http_exception.code - (http_exception.code % 100) == 500
if self.app.debug and is_server_error:
if sys.version_info < (3, 0):
exc_type, exc_value, tb = sys.exc_info()
if exc_value is exception:
reraise(exc_type, exc_value, tb)
raise exception
else:
if self.app.testing and is_server_error and isinstance(exception, Exception):
http_exception.description = '%s' % exception
return http_exception.get_response()
def abort(self, code=500, message=None, **kwargs):
'''
Properly abort the current request.
Raise a `HTTPException` for the given status `code`.
Attach any keyword arguments to the exception for later processing.
:param int code: The associated HTTP status code
:param str message: An optional details message
:param kwargs: Any additional data to pass to the error payload
:raise HTTPException:
'''
try:
flask.abort(code)
except HTTPException as e:
# JSON API specs
kwargs['errors'] = []
kwargs['errors'].append({})
kwargs['errors'][0]['detail'] = message
kwargs['errors'][0]['status'] = str(code)
kwargs['errors'][0]['title'] = str(e).split(':')[1].lstrip(' ')
e.data = kwargs
raise
def __init__(self, app, environ):
self.app = app
self.url_adapter = app.url_map.bind_to_environ(environ,
server_name=app.config['SERVER_NAME'])
self.request = app.request_class(environ)
self.session = app.open_session(self.request)
if self.session is None:
self.session = _NullSession()
self.g = _RequestGlobals()
self.flashes = None
try:
self.request.endpoint, self.request.view_args = \
self.url_adapter.match()
except HTTPException, e:
self.request.routing_exception = e
def __init__(self, app, environ):
self.app = app
self.url_adapter = app.url_map.bind_to_environ(environ)
self.request = app.request_class(environ)
# ??(session) ??:
self.session = app.open_session(self.request) # ????: session: ?????? session ??
if self.session is None:
self.session = _NullSession()
# g ??:
self.g = _RequestGlobals() # g: ?????? g ??
self.flashes = None
try:
self.request.endpoint, self.request.view_args = \
self.url_adapter.match()
except HTTPException, e:
self.request.routing_exception = e
#
# ??:
# - ?? ?????
# - ? ??????, ?? --> ??????? g, session
#
def dispatch_request(self):
"""Does the request dispatching. Matches the URL and returns the
return value of the view or error handler. This does not have to
be a response object. In order to convert the return value to a
proper response object, call :func:`make_response`.
"""
try:
endpoint, values = self.match_request() # ????
return self.view_functions[endpoint](**values)
except HTTPException, e:
handler = self.error_handlers.get(e.code)
if handler is None:
return e
return handler(e)
except Exception, e:
handler = self.error_handlers.get(500)
if self.debug or handler is None:
raise
return handler(e)
# ????
def with_error_response(logger):
def _deco(func):
@functools.wraps(func)
def _wrapper(*args, **kwargs):
try:
return func(*args, **kwargs)
except HTTPException as exc:
logger.exception("Tolerated error: %s", exc)
raise
except SQLAlchemyError as exc:
logger.exception("SQLAlchemy error: %s", exc)
raise
except Exception as exc:
logger.exception("Unknown error: %s", exc)
raise
return _wrapper
return _deco
def returns_json(f):
"""Decorator to add the content type to responses."""
@wraps(f)
def decorated_function(*args, **kwargs):
try:
r = f(*args, **kwargs)
except HTTPException as e:
# monkey-patch the headers / body to be json
headers = e.get_headers()
for header in headers:
if 'Content-Type' in header:
headers.remove(header)
headers.append(('Content-Type', 'application/json'))
e.get_headers = lambda x: headers
e.get_body = lambda x: json.dumps({"message": e.description})
raise e
if isinstance(r, tuple):
return Response(r[0], status=r[1], content_type='application/json')
else:
return Response(r, content_type='application/json')
return decorated_function
