JSONTranscoder: Add support for datetime, UUID, and binary types.

This commit is contained in:
Dave Shawley 2016-01-10 16:04:38 -05:00
parent ee8f645a51
commit 61713f9f15
2 changed files with 129 additions and 3 deletions

View file

@ -1,4 +1,10 @@
"""Bundled media type transcoders."""
"""
Bundled media type transcoders.
- :class:`.JSONTranscoder` implements JSON encoding/decoding
"""
import base64
import json
import uuid
@ -16,10 +22,16 @@ class JSONTranscoder(handlers.TextContentHandler):
If omitted, this defaults to ``utf-8``. This is passed directly to
the ``TextContentHandler`` initializer.
This JSON encoder uses :func:`json.loads` and :func:`json.dumps` to
implement JSON encoding/decoding. The :meth:`dump_object` method is
configured to handle types that the standard JSON module does not
support.
.. attribute:: dump_options
Keyword parameters that are passed to :func:`json.dumps` when
:meth:`.dumps` is called.
:meth:`.dumps` is called. By default, the :meth:`dump_object`
method is enabled as the default object hook.
.. attribute:: load_options
@ -32,7 +44,10 @@ class JSONTranscoder(handlers.TextContentHandler):
default_encoding='utf-8'):
super(JSONTranscoder, self).__init__(content_type, self.dumps,
self.loads, default_encoding)
self.dump_options = {}
self.dump_options = {
'default': self.dump_object,
'separators': (',', ':'),
}
self.load_options = {}
def dumps(self, obj):
@ -54,3 +69,46 @@ class JSONTranscoder(handlers.TextContentHandler):
"""
return json.loads(str_repr, **self.load_options)
def dump_object(self, obj):
"""
Called to encode unrecognized object.
:param object obj: the object to encode
:return: the encoded object
:raises TypeError: when `obj` cannot be encoded
This method is passed as the ``default`` keyword parameter
to :func:`json.dumps`. It provides default representations for
a number of Python language/standard library types.
+----------------------------+---------------------------------------+
| Python Type | String Format |
+----------------------------+---------------------------------------+
| :class:`bytes`, | Base64 encoded string. |
| :class:`bytearray`, | |
| :class:`memoryview` | |
+----------------------------+---------------------------------------+
| :class:`datetime.datetime` | ISO8601 formatted timestamp in the |
| | extended format including separators, |
| | milliseconds, and the timezone |
| | designator. |
+----------------------------+---------------------------------------+
| :class:`uuid.UUID` | Same as ``str(value)`` |
+----------------------------+---------------------------------------+
.. warning::
:class:`bytes` instances are treated as character strings by the
standard JSON module in Python 2.7 so the *default* object hook
is never called. In other words, :class:`bytes` values will not
be serialized as Base64 strings in Python 2.7.
"""
if isinstance(obj, uuid.UUID):
return str(obj)
if hasattr(obj, 'isoformat'):
return obj.isoformat()
if isinstance(obj, (bytes, bytearray, memoryview)):
return base64.b64encode(obj).decode('ASCII')
raise TypeError('{!r} is not JSON serializable'.format(obj))

View file

@ -1,11 +1,31 @@
import base64
import datetime
import json
import os
import sys
import unittest
import uuid
from tornado import testing
import msgpack
from sprockets.mixins.mediatype import transcoders
import examples
class UTC(datetime.tzinfo):
ZERO = datetime.timedelta(0)
def utcoffset(self, dt):
return self.ZERO
def dst(self, dt):
return self.ZERO
def tzname(self, dt):
return 'UTC'
class SendResponseTests(testing.AsyncHTTPTestCase):
def get_app(self):
@ -66,3 +86,51 @@ class GetRequestBodyTests(testing.AsyncHTTPTestCase):
headers={'Content-Type': 'application/msgpack'})
self.assertEqual(response.code, 200)
self.assertEqual(json.loads(response.body.decode('utf-8')), body)
class JSONTranscoderTests(unittest.TestCase):
def setUp(self):
super(JSONTranscoderTests, self).setUp()
self.transcoder = transcoders.JSONTranscoder()
def test_that_uuids_are_dumped_as_strings(self):
obj = {'id': uuid.uuid4()}
dumped = self.transcoder.dumps(obj)
self.assertEqual(dumped.replace(' ', ''), '{"id":"%s"}' % obj['id'])
def test_that_datetimes_are_dumped_in_isoformat(self):
obj = {'now': datetime.datetime.now()}
dumped = self.transcoder.dumps(obj)
self.assertEqual(dumped.replace(' ', ''),
'{"now":"%s"}' % obj['now'].isoformat())
def test_that_tzaware_datetimes_include_tzoffset(self):
obj = {'now': datetime.datetime.now().replace(tzinfo=UTC())}
self.assertTrue(obj['now'].isoformat().endswith('+00:00'))
dumped = self.transcoder.dumps(obj)
self.assertEqual(dumped.replace(' ', ''),
'{"now":"%s"}' % obj['now'].isoformat())
@unittest.skipIf(sys.version_info[0] == 2, 'bytes unsupported on python 2')
def test_that_bytes_are_base64_encoded(self):
bin = bytes(os.urandom(127))
dumped = self.transcoder.dumps({'bin': bin})
self.assertEqual(
dumped, '{"bin":"%s"}' % base64.b64encode(bin).decode('ASCII'))
def test_that_bytearrays_are_base64_encoded(self):
bin = bytearray(os.urandom(127))
dumped = self.transcoder.dumps({'bin': bin})
self.assertEqual(
dumped, '{"bin":"%s"}' % base64.b64encode(bin).decode('ASCII'))
def test_that_memoryviews_are_base64_encoded(self):
bin = memoryview(os.urandom(127))
dumped = self.transcoder.dumps({'bin': bin})
self.assertEqual(
dumped, '{"bin":"%s"}' % base64.b64encode(bin).decode('ASCII'))
def test_that_unhandled_objects_raise_type_error(self):
with self.assertRaises(TypeError):
self.transcoder.dumps(object())