Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
@webob.dec.wsgify
def application(req):
return 'Hello, World!!!'
@webob.dec.wsgify(RequestClass=wsgi.Request)
def __call__(self, req):
LOG.debug('Request: %s', req)
path = req.path_info
if path == '' or path[0] != '/':
path = '/' + path
path = posixpath.normpath(path)
path_tokens = path.split('/')[1:]
if path_tokens[0] == 'ec2':
path_tokens = path_tokens[1:]
if path_tokens == ['']:
resp = api.get_version_list()
return self._add_response_data(req.response, resp)
try:
@webob.dec.wsgify(RequestClass=Request)
def __call__(self, req):
"""Route the incoming request to a controller based on self.map.
If no match, return a 404.
"""
return self._router
@webob.dec.wsgify(RequestClass=wsgi.Request)
def __call__(self, req):
"""Generate a WSGI response.
Response is generated based on the exception passed to constructor.
"""
# Replace the body with fault details.
code = self.wrapped_exc.status_int
fault_name = self._fault_names.get(code, "quantumServiceFault")
fault_data = {
fault_name: {
'code': code,
'message': self.wrapped_exc.explanation}}
# 'code' is an attribute on the fault tag itself
content_type = req.best_match_content_type()
self.wrapped_exc.body = wsgi.Serializer().serialize(
fault_data, content_type)
@webob.dec.wsgify(RequestClass=webob.Request)
def __call__(self, req):
context = req.environ['synaps.context']
request_id = context.request_id
api_request = req.environ['cloudwatch.request']
result = None
try:
result = api_request.invoke(context)
except (exception.CloudwatchAPIError, exception.SynapsException) as ex:
if ex.code:
return cloudwatch_error(req, request_id, ex.code, unicode(ex))
else:
return cloudwatch_error(req, request_id, type(ex).__name__,
unicode(ex))
except Exception as ex:
env = req.environ.copy()
@webob.dec.wsgify(RequestClass=base_wsgi.Request)
def __call__(self, req):
LOG.debug("Request: %s", req)
try:
return self._proxy_request(req.remote_addr,
req.method,
req.path_info,
req.query_string,
req.body)
except Exception:
LOG.exception(_LE("Unexpected error."))
msg = _('An unknown error has occurred. '
'Please try your request again.')
explanation = six.text_type(msg)
return webob.exc.HTTPInternalServerError(explanation=explanation)
@webob.dec.wsgify
def __call__(self, req):
print(("*" * 40) + " REQUEST ENVIRON")
for key, value in req.environ.items():
print(key, "=", value)
print()
resp = req.get_response(self.application)
print(("*" * 40) + " RESPONSE HEADERS")
for (key, value) in six.iteritems(resp.headers):
print(key, "=", value)
print()
resp.app_iter = self.print_generator(resp.app_iter)
return resp
@webob.dec.wsgify(RequestClass=wsgi.Request)
def __call__(self, request):
"""Handles a call to this application.
Returns 204 if the request is acceptable to the limiter, else a 403
is returned with a relevant header indicating when the request *will*
succeed.
"""
if request.method != "POST":
raise webob.exc.HTTPMethodNotAllowed()
try:
info = dict(jsonutils.loads(request.body))
except ValueError:
raise webob.exc.HTTPBadRequest()
username = request.path_info_pop()
@webob.dec.wsgify(RequestClass=wsgi.Request)
def __call__(self, req):
if not self.has_authentication(req):
return self.authenticate(req)
user = self.get_user_by_authentication(req)
if not user:
token = req.headers["X-Auth-Token"]
msg = _("%(user)s could not be found with token '%(token)s'")
LOG.warn(msg % locals())
return faults.Fault(webob.exc.HTTPUnauthorized())
try:
account = req.headers["X-Auth-Project-Id"]
except KeyError:
# FIXME(usrleon): It needed only for compatibility
# while osapi clients don't use this header
accounts = self.auth.get_projects(user=user)
@webob.dec.wsgify
def dummy(self, req):
#TODO(gundlach)
msg = "dummy response -- please hook up __init__() to cloud.py instead"
return repr({ 'dummy': msg,
'kwargs': repr(req.environ['wsgiorg.routing_args'][1]) })