Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def __execute_request(self, func, args, req, environ):
args = self.__build_args(func, args, req, environ)
try:
result = func(**args)
except exc.HTTPException as e:
result = e
return result
def __call__(self, environ, start_response):
req = Request(environ)
action = req.params.get('action', 'view')
page = self.get_page(req.path_info)
try:
try:
meth = getattr(self, 'action_%s_%s' % (action, req.method))
except AttributeError:
raise exc.HTTPBadRequest('No such action %r' % action).exception
resp = meth(req, page)
except exc.HTTPException, e:
resp = e
return resp(environ, start_response)
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import webob.dec
import webob.exc
from quantum.api import api_common as common
from quantum.common import wsgi
class Fault(webob.exc.HTTPException):
"""Error codes for API faults"""
_fault_names = {
400: "malformedRequest",
401: "unauthorized",
420: "networkNotFound",
421: "networkInUse",
422: "networkNameExists",
430: "portNotFound",
431: "requestedStateInvalid",
432: "portInUse",
433: "portIsDown",
440: "alreadyAttached",
470: "serviceUnavailable",
471: "pluginFault"}
def object_dispatch(obj, url_path):
remainder = url_path
notfound_handlers = []
while True:
try:
obj, remainder = find_object(obj, remainder, notfound_handlers)
return obj, remainder
except HTTPException:
if not notfound_handlers:
raise HTTPNotFound().exception
name, obj, remainder = notfound_handlers.pop()
if name == 'default':
return obj, remainder
else:
obj, remainder = obj(*remainder)
continue
def _init_flow_exceptions():
"""Internal helper to initialize _flow_exceptions.
This automatically adds webob.exc.HTTPException, if it can be imported.
"""
global _flow_exceptions
_flow_exceptions = ()
add_flow_exception(datastore_errors.Rollback)
try:
from webob import exc
except ImportError:
pass
else:
add_flow_exception(exc.HTTPException)
def __call__(self, environ, start_response):
req = Request(environ)
action = req.params.get('action', 'view')
page = self.get_page(req.path_info)
try:
try:
meth = getattr(self, 'action_%s_%s' % (action, req.method))
except AttributeError:
raise exc.HTTPBadRequest('No such action %r' % action)
resp = meth(req, page)
except exc.HTTPException, e:
resp = e
return resp(environ, start_response)
def __init__(self, code, label, title, explanation, **kw):
self.code = code
self.label = label
self.title = title
self.explanation = explanation
Response.__init__(self, status='%s %s' % (self.code, self.title), **kw)
webob.exc.HTTPException.__init__(self, self.explanation, self)
def call_func(self, req, *args, **kwargs):
"""Add json_error_formatter to any webob HTTPExceptions."""
try:
return super(SdkWsgify, self).call_func(req, *args, **kwargs)
except webob.exc.HTTPException as exc:
msg = ('encounter %(error)s error') % {'error': exc}
LOG.debug(msg)
exc.json_formatter = json_error_formatter
code = exc.status_int
explanation = six.text_type(exc)
fault_data = {
'overallRC': 400,
'rc': 400,
'rs': code,
'modID': SDKWSGI_MODID,
'output': '',
'errmsg': explanation}
exc.text = six.text_type(json.dumps(fault_data))
raise exc
LOG = log.LOG
NAME = "zvm-cloud-connector"
def _find_fault(clazz, encountered=None):
if not encountered:
encountered = []
for subclass in clazz.__subclasses__():
if subclass not in encountered:
encountered.append(subclass)
for subsubclass in _find_fault(subclass, encountered):
yield subsubclass
yield subclass
class Fault(webob.exc.HTTPException):
def __init__(self, exception):
self.wrapped_exc = exception
for key, value in list(self.wrapped_exc.headers.items()):
self.wrapped_exc.headers[key] = str(value)
self.status_int = exception.status_int
@webob.dec.wsgify()
def __call__(self, req):
code = self.wrapped_exc.status_int
explanation = self.wrapped_exc.explanation
LOG.debug("Returning %(code)s to user: %(explanation)s",
{'code': code, 'explanation': explanation})
fault_data = {
# if this is not an internal redirect, run error hooks
on_error_result = None
if not internal_redirect:
on_error_result = self.handle_hooks(
self.determine_hooks(state.controller),
'on_error',
state,
e
)
# if the on_error handler returned a Response, use it.
if isinstance(on_error_result, WebObResponse):
state.response = on_error_result
else:
if not isinstance(e, exc.HTTPException):
raise
# if this is an HTTP 405, attempt to specify an Allow header
if isinstance(e, exc.HTTPMethodNotAllowed) and controller:
allowed_methods = _cfg(controller).get('allowed_methods', [])
if allowed_methods:
state.response.allow = sorted(allowed_methods)
finally:
# if this is not an internal redirect, run "after" hooks
if not internal_redirect:
self.handle_hooks(
self.determine_hooks(state.controller),
'after',
state
)