Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
async def pure_handler(request):
log.log("in pure_handler")
data = None
if request.has_body:
log.log("has body: {}".format(request.has_body))
log.log("request comes with data")
data = await request.content.read()
response = await runner.handle_request(
handle_func, constants.HTTPSTREAM,
request=request, data=data)
log.log("request execution completed")
headers = response.context().GetResponseHeaders()
response_content_type = headers.get(
constants.CONTENT_TYPE, "application/json"
)
headers.set(constants.CONTENT_TYPE, response_content_type)
kwargs = {
"headers": headers.http_raw()
}
sdata = serialize_response_data(
response.body(), response_content_type)
async def pure_handler(request):
logger.info("in pure_handler")
func_response = await runner.handle_request(
handle_code, constants.HTTPSTREAM,
headers=dict(request.headers), data=io.BytesIO(request.body))
logger.info("request execution completed")
return func_response
async def pure_handler(request):
from fdk import runner
logger.info("in pure_handler")
headers = dict(request.headers)
log_frame_header(headers)
func_response = await runner.handle_request(
handle_code, constants.HTTPSTREAM,
headers=headers, data=io.BytesIO(request.body))
logger.info("request execution completed")
headers = func_response.context().GetResponseHeaders()
status = func_response.status()
if status not in constants.FN_ENFORCED_RESPONSE_CODES:
status = constants.FN_DEFAULT_RESPONSE_CODE
return response.HTTPResponse(
headers=headers,
status=status,
content_type=headers.get(constants.CONTENT_TYPE),
body_bytes=func_response.body_bytes(),
)
async def setup_fn_call_raw(handle_func, content=None, headers=None):
if headers is None:
headers = {}
# don't decap headers, so we can test them
# (just like they come out of fdk)
return process_response(runner.handle_request(
code(handle_func), constants.HTTPSTREAM,
headers=headers, data=content,
))