python类exc()的实例源码

def _validate_input_request(self, payload):
        for key in payload:
            if " " in payload.get(key):
                LOG.debug('Input payload contain white spaces %s' %
                          str(payload))
                msg = _('%s contains white spaces') % key
                raise webob.exc.HTTPBadRequest(explanation=msg)

def ord_notifier(self, **args):
        raise webob.exc.HTTPNotFound

def __exit__(self, ex_type, ex_value, ex_traceback):
        if not ex_value:
            return True

        if isinstance(ex_value, exception.Forbidden):
            raise Fault(webob.exc.HTTPForbidden(
                        explanation=ex_value.format_message()))
        elif isinstance(ex_value, exception.VersionNotFoundForAPIMethod):
            raise
        elif isinstance(ex_value, exception.Invalid):
            raise Fault(exception.ConvertedException(
                        code=ex_value.code,
                        explanation=ex_value.format_message()))
        elif isinstance(ex_value, TypeError):
            exc_info = (ex_type, ex_value, ex_traceback)
            LOG.error('Exception handling resource: %s', ex_value,
                      exc_info=exc_info)
            raise Fault(webob.exc.HTTPBadRequest())
        elif isinstance(ex_value, Fault):
            LOG.info("Fault thrown: %s", ex_value)
            raise ex_value
        elif isinstance(ex_value, webob.exc.HTTPException):
            LOG.info("HTTP exception thrown: %s", ex_value)
            raise Fault(ex_value)

        # We didn't handle the exception
        return False

def __call__(self, request):
        """WSGI method that controls (de)serialization and method dispatch."""

        if self.support_api_request_version:
            # Set the version of the API requested based on the header
            try:
                request.set_api_version_request()
            except exception.InvalidAPIVersionString as e:
                return Fault(webob.exc.HTTPBadRequest(
                    explanation=e.format_message()))
            except exception.InvalidGlobalAPIVersion as e:
                return Fault(webob.exc.HTTPNotAcceptable(
                    explanation=e.format_message()))

        # Identify the action, its arguments, and the requested
        # content type
        action_args = self.get_action_args(request.environ)
        action = action_args.pop('action', None)

        # NOTE: we filter out InvalidContentTypes early so we
        # know everything is good from here on out.
        try:
            content_type, body = self.get_body(request)
            accept = request.best_match_content_type()
        except exception.InvalidContentType:
            msg = _("Unsupported Content-Type")
            return Fault(webob.exc.HTTPUnsupportedMediaType(explanation=msg))

        # NOTE: Splitting the function up this way allows for
        #       auditing by external tools that wrap the existing
        #       function.  If we try to audit __call__(), we can
        #       run into troubles due to the @webob.dec.wsgify()
        #       decorator.
        return self._process_stack(request, action, action_args,
                                   content_type, body, accept)

def _get_method(self, request, action, content_type, body):
        """Look up the action-specific method and its extensions."""
        # Look up the method
        try:
            if not self.controller:
                meth = getattr(self, action)
            else:
                meth = getattr(self.controller, action)
        except AttributeError:
            if (not self.wsgi_actions or
                    action not in _ROUTES_METHODS + ['action']):
                if self.controller:
                    msg = _("The request method: '%(method)s' with action: "
                            "'%(action)s' is not allowed on this "
                            "resource") % {
                                'method': request.method, 'action': action
                        }
                    raise webob.exc.HTTPMethodNotAllowed(
                        explanation=msg, body_template='${explanation}')
                # Propagate the error
                raise
        else:
            return meth, self.wsgi_extensions.get(action, [])

        if action == 'action':
            action_name = action_peek(body)
        else:
            action_name = action

        # Look up the action method
        return (self.wsgi_actions[action_name],
                self.wsgi_action_extensions.get(action_name, []))

def dispatch(self, method, request, action_args):
        """Dispatch a call to the action-specific method."""

        try:
            return method(req=request, **action_args)
        except exception.VersionNotFoundForAPIMethod:
            # We deliberately don't return any message information
            # about the exception to the user so it looks as if
            # the method is simply not implemented.
            return Fault(webob.exc.HTTPNotFound())

def manage_http_exception(code, message):
    exc = default_exceptions.get(code, webob.exc.HTTPInternalServerError)
    return exc("%s" % message)

def __init__(self, exc=None, message=None, code=None):
        details = get_exception_details(exc, message, code)
        self.message = details['message']
        self.code = details['code']
        make_log("debug", message)

def __init__(self, message, exc=None, code=400):
        super(ParseException, self).__init__(
            exc, message, code
        )

def __init__(self, message, exc=None, code=401):
        super(UserCredentialsException, self).__init__(
            exc, message, code
        )
        self.message = ("User Credentials Exception: %s "
                        % self.message)

def __init__(self, message, exc=None, code=None):
        super(ConfigurationException, self).__init__(
            exc, message, code
        )
        self.message = ("Configuration Exception: %s "
                        % self.message)

def __init__(self, exc=None, message=None, code=None):
        super(DockerException, self).__init__(
            exc, message, code
        )
        self.message = ("Error: %s "
                        % self.message)

def __call__(self, req):
        serializer = wsgi.JSONResponseSerializer()
        resp = webob.Response(request=req)
        default_webob_exc = webob.exc.HTTPInternalServerError()
        resp.status_code = self.error.get('code', default_webob_exc.code)
        serializer.default(resp, self.error)
        return resp

def __call__(self, req):
        try:
            return req.get_response(self.application)
        except Exception as exc:
            LOG.exception(exc)
            return req.get_response(Fault(self._error(exc)))

def reject(self, req, allowed_methods, *args, **kwargs):
        LOG.debug("The method %s is not allowed for this resource",
                  req.environ['REQUEST_METHOD'])
        raise webob.exc.HTTPMethodNotAllowed(
            headers=[('Allow', allowed_methods)])

def _dispatch(req):
        """Called by self._router after matching the incoming request to a route
        and putting the information into req.environ.  Either returns 404,
        501, or the routed WSGI app's response.
        """
        match = req.environ['wsgiorg.routing_args'][1]
        if not match:
            implemented_http_methods = ['GET', 'HEAD', 'POST', 'PUT',
                                        'DELETE', 'PATCH']
            if req.environ['REQUEST_METHOD'] not in implemented_http_methods:
                return webob.exc.HTTPNotImplemented()
            else:
                return webob.exc.HTTPNotFound()
        app = match['controller']
        return app

def get_content_range(self):
        """Return the `Range` in a request."""
        range_str = self.headers.get('Content-Range')
        if range_str is not None:
            range_ = webob.byterange.ContentRange.parse(range_str)
            if range_ is None:
                msg = _('Malformed Content-Range header: %s') % range_str
                raise webob.exc.HTTPBadRequest(explanation=msg)
            return range_

def test_413_fault_json(self):
        # Test fault serialized to JSON via file-extension and/or header.
        requests = [
            webob.Request.blank('/.json'),
            webob.Request.blank('/', headers={"Accept": "application/json"}),
        ]

        for request in requests:
            exc = webob.exc.HTTPRequestEntityTooLarge
            # NOTE(aloga): we intentionally pass an integer for the
            # 'Retry-After' header. It should be then converted to a str
            fault = wsgi.Fault(exc(explanation='sorry',
                        headers={'Retry-After': 4}))
            response = request.get_response(fault)

            expected = {
                "overLimit": {
                    "message": "sorry",
                    "code": 413,
                    "retryAfter": "4",
                },
            }
            actual = jsonutils.loads(response.body)

            self.assertEqual(response.content_type, "application/json")
            self.assertEqual(expected, actual)

def test_429_fault_json(self):
        # Test fault serialized to JSON via file-extension and/or header.
        requests = [
            webob.Request.blank('/.json'),
            webob.Request.blank('/', headers={"Accept": "application/json"}),
        ]

        for request in requests:
            exc = webob.exc.HTTPTooManyRequests
            # NOTE(aloga): we intentionally pass an integer for the
            # 'Retry-After' header. It should be then converted to a str
            fault = wsgi.Fault(exc(explanation='sorry',
                        headers={'Retry-After': 4}))
            response = request.get_response(fault)

            expected = {
                "overLimit": {
                    "message": "sorry",
                    "code": 429,
                    "retryAfter": "4",
                },
            }
            actual = jsonutils.loads(response.body)

            self.assertEqual(response.content_type, "application/json")
            self.assertEqual(expected, actual)

def test_fault_has_status_int(self):
        # Ensure the status_int is set correctly on faults.
        fault = wsgi.Fault(webob.exc.HTTPBadRequest(explanation='what?'))
        self.assertEqual(fault.status_int, 400)