2019-02-04 18:58:38 +00:00
|
|
|
"""Helper functions for cloud components."""
|
|
|
|
from typing import Any, Dict
|
|
|
|
|
2019-12-08 17:01:12 +00:00
|
|
|
from aiohttp import payload, web
|
2019-02-04 18:58:38 +00:00
|
|
|
|
|
|
|
|
|
|
|
def aiohttp_serialize_response(response: web.Response) -> Dict[str, Any]:
|
|
|
|
"""Serialize an aiohttp response to a dictionary."""
|
2019-05-14 09:59:27 +00:00
|
|
|
body = response.body
|
|
|
|
|
|
|
|
if body is None:
|
|
|
|
pass
|
|
|
|
elif isinstance(body, payload.StringPayload):
|
|
|
|
# pylint: disable=protected-access
|
|
|
|
body = body._value.decode(body.encoding)
|
|
|
|
elif isinstance(body, bytes):
|
2019-07-31 19:25:30 +00:00
|
|
|
body = body.decode(response.charset or "utf-8")
|
2019-05-14 09:59:27 +00:00
|
|
|
else:
|
|
|
|
raise ValueError("Unknown payload encoding")
|
|
|
|
|
2019-07-31 19:25:30 +00:00
|
|
|
return {"status": response.status, "body": body, "headers": dict(response.headers)}
|