You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
86 lines
2.9 KiB
Python
86 lines
2.9 KiB
Python
import datetime as dt
|
|
import decimal
|
|
import json
|
|
import typing
|
|
|
|
from apistar import types, exceptions
|
|
from apistar.http import JSONResponse, Response, StrMapping, StrPairs, MutableHeaders
|
|
from werkzeug.http import HTTP_STATUS_CODES
|
|
|
|
|
|
class ExtJSONResponse(JSONResponse):
|
|
"""JSON Response with support for ISO 8601 datetime serialization and Decimal to float casting"""
|
|
|
|
def default(self, obj: typing.Any) -> typing.Any:
|
|
if isinstance(obj, types.Type):
|
|
return dict(obj)
|
|
if isinstance(obj, dt.datetime):
|
|
return obj.isoformat()
|
|
elif isinstance(obj, decimal.Decimal):
|
|
return float(obj)
|
|
error = "Object of type '%s' is not JSON serializable."
|
|
return TypeError(error % type(obj).__name_)
|
|
|
|
|
|
class MetaJSONResponse(Response):
|
|
"""A JSONResponse that returns a meta, data, error on response"""
|
|
media_type = 'application/json'
|
|
charset = None
|
|
options = {
|
|
'ensure_ascii': False,
|
|
'allow_nan': False,
|
|
'indent': None,
|
|
'separators': (',', ':'),
|
|
}
|
|
|
|
def __init__(self,
|
|
content: typing.Any,
|
|
status_code: int = 200,
|
|
headers: typing.Union[StrMapping, StrPairs] = None,
|
|
exc_info=None,
|
|
meta: typing.Any = None) -> None:
|
|
self.status_code = status_code
|
|
self.meta = self._build_meta(meta)
|
|
self._content = content
|
|
self.content = self.render(content)
|
|
self.headers = MutableHeaders(headers)
|
|
self.set_default_headers()
|
|
self.exc_info = exc_info
|
|
|
|
def render(self, content: typing.Any) -> bytes:
|
|
"""Builds a JSON response containing meta data. If the content is an `apistar.execption.HTTPException`
|
|
then it will return the execption reason code"""
|
|
error = {}
|
|
if isinstance(content, exceptions.HTTPException):
|
|
error["reason"] = content.detail
|
|
|
|
options = {'default': self.default}
|
|
options.update(self.options)
|
|
response = dict(meta=self.meta)
|
|
|
|
if error:
|
|
response.update(dict(error=error))
|
|
return json.dumps(response, **options).encode('utf-8')
|
|
|
|
response.update(dict(data=content))
|
|
return json.dumps(response, **options).encode('utf-8')
|
|
|
|
def default(self, obj: typing.Any) -> typing.Any:
|
|
if isinstance(obj, types.Type):
|
|
return dict(obj)
|
|
if isinstance(obj, dt.datetime):
|
|
return obj.isoformat()
|
|
elif isinstance(obj, decimal.Decimal):
|
|
return float(obj)
|
|
error = "Object of type '%s' is not JSON serializable."
|
|
return TypeError(error % type(obj).__name_)
|
|
|
|
def _build_meta(self, meta):
|
|
_meta = {
|
|
"status": self.status_code,
|
|
"message": HTTP_STATUS_CODES.get(self.status_code)
|
|
}
|
|
if meta is None:
|
|
return _meta
|
|
return {**_meta, **meta}
|