Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def aiohttp_multiple_methods_get(name, **kwargs):
data = {'method': 'get'}
return ConnexionResponse(
body=data
)
def aiohttp_str_response():
return ConnexionResponse(
body='str response'
)
def aiohttp_multiple_methods_post(name, **kwargs):
data = {'method': 'post'}
return ConnexionResponse(
body=data
)
def aiohttp_bytes_response():
return ConnexionResponse(
body=b'bytes response'
)
def _framework_to_connexion_response(cls, response, mimetype):
""" Cast framework response class to ConnexionResponse used for schema validation """
return ConnexionResponse(
status_code=response.status,
mimetype=mimetype,
content_type=response.content_type,
headers=response.headers,
body=response.body
)
code will be set to 200 and no headers will be added.
If the returned object is a flask.Response then it will just
pass the information needed to recreate it.
:type operation_handler_result: flask.Response | (flask.Response, int) | (flask.Response, int, dict)
:rtype: ConnexionRequest
"""
logger.debug('Getting data and status code',
extra={
'data': response,
'data_type': type(response),
'url': flask.request.url
})
if isinstance(response, ConnexionResponse):
flask_response = cls._get_flask_response_from_connexion(response, mimetype)
else:
flask_response = cls._get_flask_response(response, mimetype)
logger.debug('Got data and status code (%d)',
flask_response.status_code,
extra={
'data': response,
'datatype': type(response),
'url': flask.request.url
})
return flask_response
2. handlers that return a code that is not in the swagger spec.
In both cases, the exception would be punted here, and we return this very generic error that also happens to
bypass all validation.
"""
problem = {
'status': requests.codes.server_error,
'code': "unhandled_exception",
'title': str(exception),
'stacktrace': traceback.format_exc(),
}
if isinstance(exception, (OAuthProblem, OAuthResponseProblem, OAuthScopeProblem, Forbidden)):
problem['status'] = exception.code
problem['code'] = exception.__class__.__name__
problem['title'] = exception.description
return FlaskApi.get_response(ConnexionResponse(
status_code=problem['status'],
mimetype="application/problem+json",
content_type="application/problem+json",
body=problem,
))
async def get_response(cls, response, mimetype=None, request=None):
"""Get response.
This method is used in the lifecycle decorators
:rtype: aiohttp.web.Response
"""
while asyncio.iscoroutine(response):
response = await response
logger.debug('Getting data', extra={'data': response})
if isinstance(response, ConnexionResponse):
response = cls._get_aiohttp_response_from_connexion(response,
mimetype)
elif isinstance(response, tuple):
response = cls._json_response(response[0], response[1])
else:
response = cls._json_response(response)
url = str(request.url) if request else ''
logger.debug('Got data and status code (%d)', response.status,
extra={'data': response.body, 'url': url})
return response
result. Status Code and Headers for response. If only body
data is returned by the endpoint function, then the status
code will be set to 200 and no headers will be added.
"""
logger.debug('Getting data and status code',
extra={
'data': response,
'data_type': type(response),
'url': 'TODO',
})
falcon_response = request.context.falcon_response
if response == falcon_response:
pass
elif isinstance(response, ConnexionResponse):
falcon_response.status = cls._get_status(response.status_code)
falcon_response.content_type = response.content_type or response.mimetype
falcon_response.body = response.body
falcon_response.set_headers(response.headers or {})
else:
cls._set_falcon_response(response, mimetype, falcon_response)
logger.debug('Got data and status code (%s)',
falcon_response.status,
extra={
'data': response,
'datatype': type(response),
'url': 'TODO'
})
return falcon_response
:param response: A response to cast (tuple, framework response, etc).
:param mimetype: The response mimetype.
:type mimetype: Union[None, str]
:param extra_context: dict of extra details, like url, to include in logs
:type extra_context: Union[None, dict]
"""
if extra_context is None:
extra_context = {}
logger.debug('Getting data and status code',
extra={
'data': response,
'data_type': type(response),
**extra_context
})
if isinstance(response, ConnexionResponse):
framework_response = cls._connexion_to_framework_response(response, mimetype, extra_context)
else:
framework_response = cls._response_from_handler(response, mimetype, extra_context)
logger.debug('Got framework response',
extra={
'response': framework_response,
'response_type': type(framework_response),
**extra_context
})
return framework_response