Source code for yggdrasil.metaschema.encoder

import importlib
import json as stdjson
import yaml
import rapidjson as json
from yggdrasil import tools
_json_encoder = json.Encoder
_json_decoder = json.Decoder


[docs]def indent_char2int(indent): r"""Convert a character indent into a number of spaces that should be used. Tabs are set to be equivalent to 4 spaces. Args: indent (str): String indent. Returns: int: Number of whitespaces that is equivalent to the provided string. """ if isinstance(indent, str): indent = len(indent.replace('\t', ' ')) return indent
[docs]def string2import(s): r"""Import a function/class based on its representation as a string. Args: s (str): String that may or may not contain a represetnation of an importable class or function. Returns: str, class, function: Imported class/function if one exists, original string if not. """ pkg_mod = s.split(u':') if (len(pkg_mod) == 2) and (not s.startswith('http')) and (' ' not in s): try: mod = importlib.import_module(pkg_mod[0]) s = getattr(mod, pkg_mod[1]) except (ImportError, AttributeError): # pragma: debug pass return s
[docs]class JSONReadableEncoder(stdjson.JSONEncoder): r"""Encoder class for Ygg messages."""
[docs] def default(self, o): # pragma: no cover r"""Encoder that allows for expansion types.""" from yggdrasil.metaschema import MetaschemaTypeError from yggdrasil.metaschema.datatypes import encode_data_readable try: return encode_data_readable(o) except MetaschemaTypeError: raise TypeError("Cannot encode %s" % o)
[docs]class JSONEncoder(_json_encoder): r"""Encoder class for Ygg messages."""
[docs] def default(self, o): r"""Encoder that allows for expansion types.""" from yggdrasil.metaschema import MetaschemaTypeError from yggdrasil.metaschema.datatypes import encode_data try: return encode_data(o) except MetaschemaTypeError: raise TypeError("Cannot encode %s" % o)
[docs]class JSONDecoder(_json_decoder): r"""Decoder class for Ygg messages."""
[docs] def string(self, s): r"""Try to parse string with class.""" # TODO: Do this dynamically for classes based on an attribute return string2import(s)
[docs]def encode_json(obj, fd=None, indent=None, sort_keys=True, **kwargs): r"""Encode a Python object in JSON format. Args: obj (object): Python object to encode. fd (file, optional): File descriptor for file that encoded object should be written to. Defaults to None and string is returned. indent (int, str, optional): Indentation for new lines in encoded string. Defaults to None. sort_keys (bool, optional): If True, the keys will be output in sorted order. Defaults to True. **kwargs: Additional keyword arguments are passed to json.dumps. Returns: str, bytes: Encoded object. """ if (indent is None) and (fd is not None): indent = '\t' # Character indents not allowed in Python 2 json indent = indent_char2int(indent) kwargs['indent'] = indent kwargs['sort_keys'] = sort_keys if 'cls' in kwargs: kwargs.setdefault('default', kwargs.pop('cls')().default) else: kwargs.setdefault('default', JSONEncoder().default) if fd is None: return tools.str2bytes(json.dumps(obj, **kwargs)) else: return json.dump(obj, fd, **kwargs)
[docs]def decode_json(msg, **kwargs): r"""Decode a Python object from a JSON serialization. Args: msg (str): JSON serialization to decode. **kwargs: Additional keyword arguments are passed to json.loads. Returns: object: Deserialized Python object. """ if isinstance(msg, (str, bytes)): msg_decode = tools.bytes2str(msg) func_decode = json.loads else: msg_decode = msg func_decode = json.load func_decode = JSONDecoder() return func_decode(msg_decode, **kwargs)
[docs]def encode_yaml(obj, fd=None, indent=None, sorted_dict_type=None, **kwargs): r"""Encode a Python object in YAML format. Args: obj (object): Python object to encode. fd (file, optional): File descriptor for file that encoded object should be written to. Defaults to None and string is returned. indent (int, str, optional): Indentation for new lines in encoded string. Defaults to None. **kwargs: Additional keyword arguments are passed to yaml.dump. Returns: str, bytes: Encoded object. """ from yggdrasil.metaschema.datatypes import encode_data_readable if (indent is None) and (fd is not None): indent = '\t' indent = indent_char2int(indent) kwargs['indent'] = indent if fd is not None: assert 'stream' not in kwargs kwargs['stream'] = fd if sorted_dict_type is not None: class OrderedDumper(kwargs.get('Dumper', yaml.SafeDumper)): pass def _dict_representer(dumper, data): return dumper.represent_mapping( yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG, data.items()) if not isinstance(sorted_dict_type, list): sorted_dict_type = [sorted_dict_type] for x in sorted_dict_type: OrderedDumper.add_representer(x, _dict_representer) kwargs['Dumper'] = OrderedDumper return yaml.dump(encode_data_readable(obj), **kwargs)
[docs]def decode_yaml(msg, sorted_dict_type=None, **kwargs): r"""Decode a Python object from a YAML serialization. Args: msg (str): YAML serialization to decode. sorted_dict_type (type, optional): Class that should be used to contain mapping objects while preserving order. Defaults to None and is ignored. **kwargs: Additional keyword arguments are passed to yaml.load. Returns: object: Deserialized Python object. """ class OrderedLoader(kwargs.get('Loader', yaml.Loader)): pass if sorted_dict_type is not None: def construct_mapping(loader, node): loader.flatten_mapping(node) return sorted_dict_type(loader.construct_pairs(node)) OrderedLoader.add_constructor( yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG, construct_mapping) def construct_scalar(loader, node): out = loader.construct_scalar(node) out = string2import(out) return out OrderedLoader.add_constructor( yaml.resolver.BaseResolver.DEFAULT_SCALAR_TAG, construct_scalar) kwargs['Loader'] = OrderedLoader out = yaml.load(msg, **kwargs) return out