from don import binary
from don import string
-from don.constants import *
-from don._shared import TaggedObject
def binary_to_string(b):
return string.serialize(binary.deserialize(b))
import collections
-from don import constants
-
-_TYPES_TO_TAGS = {
- int: constants.DEFAULT_INTEGER_ENCODING,
- float: constants.DEFAULT_DECIMAL_ENCODING,
- bytes: constants.BINARY,
- str: constants.DEFAULT_STRING_ENCODING,
- list: constants.LIST,
- dict: constants.DICTIONARY,
- collections.OrderedDict: constants.DICTIONARY,
-}
-
-TaggedObject = collections.namedtuple('TaggedObject', ['tag', 'value'])
-
-def _tag(o):
- if isinstance(o, TaggedObject):
- return o
-
- if o is None:
- return TaggedObject(tag = constants.VOID, value = o)
-
- if o is True:
- return TaggedObject(tag = constants.TRUE, value = o)
-
- if o is False:
- return TaggedObject(tag = constants.FALSE, value = o)
-
- return TaggedObject(tag = _TYPES_TO_TAGS[type(o)], value = o)
+from don import tags
ParseResult = collections.namedtuple(
'ParseResult',
import collections
import struct
-from don import constants, _shared
+from don import tags, _shared
def _binary_serialize_tag_only_type(o):
return b''
def _binary_serialize_list(items):
# TODO Enforce that items are all the same type
- items = [_shared._tag(i) for i in items]
+ items = [tags._tag(i) for i in items]
if len(items) == 0:
- item_tag = constants.VOID
+ item_tag = tags.VOID
else:
item_tag = items[0].tag
item_length = 0
serialized = b''
- key_serializer = _BINARY_SERIALIZERS[constants.UTF8]
+ key_serializer = _BINARY_SERIALIZERS[tags.UTF8]
for key, value in d.items():
assert isinstance(key, str)
return struct.pack('!II', byte_length, item_length) + serialized
_BINARY_SERIALIZERS = {
- constants.VOID: _binary_serialize_tag_only_type,
- constants.TRUE: _binary_serialize_tag_only_type,
- constants.FALSE: _binary_serialize_tag_only_type,
- constants.INT8: _pack_format_string_to_binary_serializer('!b'),
- constants.INT16: _pack_format_string_to_binary_serializer('!h'),
- constants.INT32: _pack_format_string_to_binary_serializer('!i'),
- constants.FLOAT: _pack_format_string_to_binary_serializer('!f'),
- constants.DOUBLE: _pack_format_string_to_binary_serializer('!d'),
- constants.BINARY: _encoder_to_binary_serializer(lambda b: b),
- constants.UTF8: _encoder_to_binary_serializer(lambda s: s.encode('utf-8')),
- constants.UTF16: _encoder_to_binary_serializer(lambda s: s.encode('utf-16')),
- constants.UTF32: _encoder_to_binary_serializer(lambda s: s.encode('utf-32')),
- constants.LIST: _binary_serialize_list,
- constants.DICTIONARY: _binary_serialize_dict,
+ tags.VOID: _binary_serialize_tag_only_type,
+ tags.TRUE: _binary_serialize_tag_only_type,
+ tags.FALSE: _binary_serialize_tag_only_type,
+ tags.INT8: _pack_format_string_to_binary_serializer('!b'),
+ tags.INT16: _pack_format_string_to_binary_serializer('!h'),
+ tags.INT32: _pack_format_string_to_binary_serializer('!i'),
+ tags.FLOAT: _pack_format_string_to_binary_serializer('!f'),
+ tags.DOUBLE: _pack_format_string_to_binary_serializer('!d'),
+ tags.BINARY: _encoder_to_binary_serializer(lambda b: b),
+ tags.UTF8: _encoder_to_binary_serializer(lambda s: s.encode('utf-8')),
+ tags.UTF16: _encoder_to_binary_serializer(lambda s: s.encode('utf-16')),
+ tags.UTF32: _encoder_to_binary_serializer(lambda s: s.encode('utf-32')),
+ tags.LIST: _binary_serialize_list,
+ tags.DICTIONARY: _binary_serialize_dict,
}
def serialize(o):
- o = _shared._tag(o)
+ o = tags._tag(o)
return struct.pack('!B', o.tag) + _BINARY_SERIALIZERS[o.tag](o.value)
_BYTE_SIZES_TO_UNPACK_FORMATS = {
)
def dictionary_parser(source):
- key_parser = _TAGS_TO_PARSERS[constants.UTF8]
+ key_parser = _TAGS_TO_PARSERS[tags.UTF8]
byte_length, item_length = struct.unpack('!II', source[:8])
source = source[8:]
_TAGS_TO_PARSERS = {
- constants.VOID: lambda r: _shared.ParseResult(True, None, r),
- constants.TRUE: lambda r: _shared.ParseResult(True, True, r),
- constants.FALSE: lambda r: _shared.ParseResult(True, False, r),
- constants.INT8: make_integer_parser(1),
- constants.INT16: make_integer_parser(2),
- constants.INT32: make_integer_parser(4),
- constants.INT64: make_integer_parser(8),
- constants.DOUBLE: binary64_parser,
- constants.BINARY: make_string_parser(lambda b : b),
- constants.UTF8: make_string_parser(lambda b : b.decode('utf-8')),
- constants.UTF16: make_string_parser(lambda b : b.decode('utf-16')),
- constants.UTF32: make_string_parser(lambda b : b.decode('utf-32')),
- constants.LIST: _list_parser,
- constants.DICTIONARY: dictionary_parser,
+ tags.VOID: lambda r: _shared.ParseResult(True, None, r),
+ tags.TRUE: lambda r: _shared.ParseResult(True, True, r),
+ tags.FALSE: lambda r: _shared.ParseResult(True, False, r),
+ tags.INT8: make_integer_parser(1),
+ tags.INT16: make_integer_parser(2),
+ tags.INT32: make_integer_parser(4),
+ tags.INT64: make_integer_parser(8),
+ tags.DOUBLE: binary64_parser,
+ tags.BINARY: make_string_parser(lambda b : b),
+ tags.UTF8: make_string_parser(lambda b : b.decode('utf-8')),
+ tags.UTF16: make_string_parser(lambda b : b.decode('utf-16')),
+ tags.UTF32: make_string_parser(lambda b : b.decode('utf-32')),
+ tags.LIST: _list_parser,
+ tags.DICTIONARY: dictionary_parser,
}
def _object_parser(source):
+++ /dev/null
-VOID = 0x00
-TRUE = 0x01
-FALSE = 0x02
-BOOL = (TRUE, FALSE)
-INT8 = 0x10
-INT16 = 0x11
-INT32 = 0x12
-INT64 = 0x13
-FLOAT = 0x20
-DOUBLE = 0x21
-BINARY = 0x30
-UTF8 = 0x31
-UTF16 = 0x32
-UTF32 = 0x33
-LIST = 0x40
-DICTIONARY = 0x41
-
-DEFAULT_INTEGER_ENCODING = INT32
-DEFAULT_DECIMAL_ENCODING = DOUBLE
-DEFAULT_STRING_ENCODING = UTF8
import binascii
-from don import constants, _shared
+from don import tags, _shared
def _integer_size_to_string_serializer(integer_size):
minimum = -(2 ** (integer_size - 1))
return '{ ' + ', '.join(map(serialize_kvp, d.items())) + ' }'
_STRING_SERIALIZERS = {
- constants.VOID: lambda o: 'null',
- constants.TRUE: lambda o: 'true',
- constants.FALSE: lambda o: 'false',
- constants.INT8: _integer_size_to_string_serializer(8),
- constants.INT16: _integer_size_to_string_serializer(16),
- constants.INT32: _integer_size_to_string_serializer(32),
- constants.INT64: _integer_size_to_string_serializer(64),
- constants.FLOAT: _serialize_float,
- constants.DOUBLE: _serialize_double,
- constants.BINARY: _serialize_binary,
- constants.UTF8: _utf_encoding_to_serializer('utf8'),
- constants.UTF16: _utf_encoding_to_serializer('utf16'),
- constants.UTF32: _utf_encoding_to_serializer('utf32'),
- constants.LIST: _string_serialize_list,
- constants.DICTIONARY: _string_serialize_dictionary,
+ tags.VOID: lambda o: 'null',
+ tags.TRUE: lambda o: 'true',
+ tags.FALSE: lambda o: 'false',
+ tags.INT8: _integer_size_to_string_serializer(8),
+ tags.INT16: _integer_size_to_string_serializer(16),
+ tags.INT32: _integer_size_to_string_serializer(32),
+ tags.INT64: _integer_size_to_string_serializer(64),
+ tags.FLOAT: _serialize_float,
+ tags.DOUBLE: _serialize_double,
+ tags.BINARY: _serialize_binary,
+ tags.UTF8: _utf_encoding_to_serializer('utf8'),
+ tags.UTF16: _utf_encoding_to_serializer('utf16'),
+ tags.UTF32: _utf_encoding_to_serializer('utf32'),
+ tags.LIST: _string_serialize_list,
+ tags.DICTIONARY: _string_serialize_dictionary,
}
def serialize(o):
- o = _shared._tag(o)
+ o = tags._tag(o)
return _STRING_SERIALIZERS[o.tag](o.value)
--- /dev/null
+import collections
+
+VOID = 0x00
+TRUE = 0x01
+FALSE = 0x02
+BOOL = (TRUE, FALSE)
+INT8 = 0x10
+INT16 = 0x11
+INT32 = 0x12
+INT64 = 0x13
+FLOAT = 0x20
+DOUBLE = 0x21
+BINARY = 0x30
+UTF8 = 0x31
+UTF16 = 0x32
+UTF32 = 0x33
+LIST = 0x40
+DICTIONARY = 0x41
+
+DEFAULT_INTEGER_ENCODING = INT32
+DEFAULT_DECIMAL_ENCODING = DOUBLE
+DEFAULT_STRING_ENCODING = UTF8
+
+TaggedObject = collections.namedtuple('TaggedObject', ['tag', 'value'])
+
+_TYPES_TO_TAGS = {
+ int: DEFAULT_INTEGER_ENCODING,
+ float: DEFAULT_DECIMAL_ENCODING,
+ bytes: BINARY,
+ str: DEFAULT_STRING_ENCODING,
+ list: LIST,
+ dict: DICTIONARY,
+ collections.OrderedDict: DICTIONARY,
+}
+
+def _tag(o):
+ if isinstance(o, TaggedObject):
+ return o
+
+ if o is None:
+ return TaggedObject(tag = VOID, value = o)
+
+ if o is True:
+ return TaggedObject(tag = TRUE, value = o)
+
+ if o is False:
+ return TaggedObject(tag = FALSE, value = o)
+
+ return TaggedObject(tag = _TYPES_TO_TAGS[type(o)], value = o)
import collections
import unittest
-import don
-from don import binary, string
+from don import binary, string, tags
class TestBinarySerialize(unittest.TestCase):
def test_serializes_null(self):
self.assertEqual(string.serialize(False), 'false')
def test_serializes_int8(self):
- self.assertEqual(string.serialize(don.TaggedObject(don.INT8, 1)), '1i8')
- self.assertEqual(string.serialize(don.TaggedObject(don.INT8, -1)), '-1i8')
- self.assertEqual(string.serialize(don.TaggedObject(don.INT8, 42)), '42i8')
+ self.assertEqual(string.serialize(tags.TaggedObject(tags.INT8, 1)), '1i8')
+ self.assertEqual(string.serialize(tags.TaggedObject(tags.INT8, -1)), '-1i8')
+ self.assertEqual(string.serialize(tags.TaggedObject(tags.INT8, 42)), '42i8')
def test_serializes_int16(self):
- self.assertEqual(string.serialize(don.TaggedObject(don.INT16, 1)), '1i16')
- self.assertEqual(string.serialize(don.TaggedObject(don.INT16, -1)), '-1i16')
- self.assertEqual(string.serialize(don.TaggedObject(don.INT16, 42)), '42i16')
+ self.assertEqual(string.serialize(tags.TaggedObject(tags.INT16, 1)), '1i16')
+ self.assertEqual(string.serialize(tags.TaggedObject(tags.INT16, -1)), '-1i16')
+ self.assertEqual(string.serialize(tags.TaggedObject(tags.INT16, 42)), '42i16')
def test_serializes_int32(self):
- self.assertEqual(string.serialize(don.TaggedObject(don.INT32, 1)), '1i32')
- self.assertEqual(string.serialize(don.TaggedObject(don.INT32, -1)), '-1i32')
- self.assertEqual(string.serialize(don.TaggedObject(don.INT32, 42)), '42i32')
+ self.assertEqual(string.serialize(tags.TaggedObject(tags.INT32, 1)), '1i32')
+ self.assertEqual(string.serialize(tags.TaggedObject(tags.INT32, -1)), '-1i32')
+ self.assertEqual(string.serialize(tags.TaggedObject(tags.INT32, 42)), '42i32')
def test_serializes_int64(self):
- self.assertEqual(string.serialize(don.TaggedObject(don.INT64, 1)), '1i64')
- self.assertEqual(string.serialize(don.TaggedObject(don.INT64, -1)), '-1i64')
- self.assertEqual(string.serialize(don.TaggedObject(don.INT64, 42)), '42i64')
+ self.assertEqual(string.serialize(tags.TaggedObject(tags.INT64, 1)), '1i64')
+ self.assertEqual(string.serialize(tags.TaggedObject(tags.INT64, -1)), '-1i64')
+ self.assertEqual(string.serialize(tags.TaggedObject(tags.INT64, 42)), '42i64')
def test_serializes_float(self):
- self.assertEqual(string.serialize(don.TaggedObject(don.FLOAT, 1.0)), '1.0f')
+ self.assertEqual(string.serialize(tags.TaggedObject(tags.FLOAT, 1.0)), '1.0f')
def test_serializes_double(self):
- self.assertEqual(string.serialize(don.TaggedObject(don.DOUBLE, 1.0)), '1.0d')
+ self.assertEqual(string.serialize(tags.TaggedObject(tags.DOUBLE, 1.0)), '1.0d')
def test_serializes_binary(self):
- self.assertEqual(string.serialize(don.TaggedObject(don.BINARY, b'\xde\xad\xbe\xef')), '"deadbeef"b')
+ self.assertEqual(string.serialize(tags.TaggedObject(tags.BINARY, b'\xde\xad\xbe\xef')), '"deadbeef"b')
def test_serializes_utf8(self):
- self.assertEqual(string.serialize(don.TaggedObject(don.UTF8, 'Hello, world')), '"Hello, world"utf8')
+ self.assertEqual(string.serialize(tags.TaggedObject(tags.UTF8, 'Hello, world')), '"Hello, world"utf8')
def test_serializes_utf16(self):
- self.assertEqual(string.serialize(don.TaggedObject(don.UTF16, 'Hello, world')), '"Hello, world"utf16')
+ self.assertEqual(string.serialize(tags.TaggedObject(tags.UTF16, 'Hello, world')), '"Hello, world"utf16')
def test_serializes_utf32(self):
- self.assertEqual(string.serialize(don.TaggedObject(don.UTF32, 'Hello, world')), '"Hello, world"utf32')
+ self.assertEqual(string.serialize(tags.TaggedObject(tags.UTF32, 'Hello, world')), '"Hello, world"utf32')
def test_serializes_list(self):
- self.assertEqual(string.serialize(don.TaggedObject(don.LIST, [1,2,3])), '[1i32, 2i32, 3i32]')
+ self.assertEqual(string.serialize(tags.TaggedObject(tags.LIST, [1,2,3])), '[1i32, 2i32, 3i32]')
def test_serializes_dictionary(self):
self.assertEqual(
- string.serialize(don.TaggedObject(don.DICTIONARY, collections.OrderedDict([
+ string.serialize(tags.TaggedObject(tags.DICTIONARY, collections.OrderedDict([
('foo', 1),
('bar', 'baz'),
]))),