Source code for yggdrasil.serialize.JSONSerialize

from yggdrasil.metaschema.encoder import (
    indent_char2int, encode_json, decode_json,
    JSONReadableEncoder)
from yggdrasil.serialize.SerializeBase import SerializeBase


[docs]class JSONSerialize(SerializeBase): r"""Class for serializing a python object into a bytes message using JSON. Args: indent (str, int, optional): String or number of spaces that should be used to indent each level within the seiralized structure. Defaults to '\t'. sort_keys (bool, optional): If True, the serialization of dictionaries will be in key sorted order. Defaults to True. """ _seritype = 'json' _schema_subtype_description = ('Serializes Python objects using the JSON ' 'standard.') _schema_properties = { 'indent': {'type': ['string', 'int'], 'default': '\t'}, 'sort_keys': {'type': 'boolean', 'default': True}} default_datatype = {'type': 'object'} concats_as_str = False
[docs] def func_serialize(self, args): r"""Serialize a message. Args: args (obj): Python object to be serialized. Returns: bytes, str: Serialized message. """ return encode_json(args, indent=self.indent, cls=JSONReadableEncoder)
[docs] def func_deserialize(self, msg): r"""Deserialize a message. Args: msg (str, bytes): Message to be deserialized. Returns: obj: Deserialized Python object. """ return decode_json(msg)
[docs] @classmethod def concatenate(cls, objects, **kwargs): r"""Concatenate objects to get object that would be recieved if the concatenated serialization were deserialized. Args: objects (list): Objects to be concatenated. **kwargs: Additional keyword arguments are ignored. Returns: list: Set of objects that results from concatenating those provided. """ if all([isinstance(x, dict) for x in objects]): total = dict() for x in objects: total.update(x) out = [total] elif all([isinstance(x, list) for x in objects]): total = list() for x in objects: total += x out = [total] else: # Adding additional list then causes set of objects to be serialized # as a JSON array out = [objects] return out
[docs] @classmethod def get_testing_options(cls, **kwargs): r"""Method to return a dictionary of testing options for this class. Returns: dict: Dictionary of variables to use for testing. """ # iobj = {'a': ['b', int(1), float(1.0)], 'c': {'z': 'hello'}} iobj1 = {'a': ['b', int(1), float(1.0)], 'c': {'z': 'hello'}} iobj2 = {'d': 'new field'} # iobj3 = int(2) # iobj4 = [float(2.0)] out = {'kwargs': {}, 'empty': {}, 'dtype': None, 'extra_kwargs': {}, 'objects': [iobj1, iobj2], # , iobj3, iobj4], 'typedef': {'type': 'object'}} out['contents'] = (b'{\n\t"a": [\n\t\t"b",\n\t\t1,\n\t\t1.0\n\t],' b'\n\t"c": {\n\t\t"z": "hello"\n\t},' b'\n\t"d": "new field"\n}') out['concatenate'] = [([{'a': 1}, {'b': 2}], [{'a': 1, 'b': 2}]), ([['a'], ['b']], [['a', 'b']]), ([['a'], {'b': 2}], [[['a'], {'b': 2}]])] # Version that allows for list concatentation # out['contents'] = (b'[\n\t' # b'{\n\t\t"a": [\n\t\t\t"b",\n\t\t\t1,\n\t\t\t1.0\n\t\t],' # b'\n\t\t"c": {\n\t\t\t"z": "hello"\n\t\t},' # b'\n\t\t"d": "new field"\n\t},' # b'\n\t2,\n\t2.0\n]') tab_rep = indent_char2int('\t') * b' ' out['contents'] = out['contents'].replace(b'\t', tab_rep) return out