Source code for yggdrasil.metaschema.datatypes.ContainerMetaschemaType
from yggdrasil.metaschema.datatypes import (
get_type_class, complete_typedef, encode_data, encode_data_readable)
from yggdrasil.metaschema.datatypes.MetaschemaType import MetaschemaType
[docs]class ContainerMetaschemaType(MetaschemaType):
r"""Type associated with a container of subtypes."""
name = 'container'
description = 'A container of other types.'
python_types = []
_container_type = None
_json_type = None
_json_property = None
def __init__(self, *args, **kwargs):
self._typecls = self._container_type()
super(ContainerMetaschemaType, self).__init__(*args, **kwargs)
@classmethod
def _iterate(cls, container):
r"""Iterate over the contents of the container. Each element returned
should be a tuple including an index and a value.
Args:
container (obj): Object to be iterated over.
Returns:
iterator: Iterator over elements in the container.
"""
raise NotImplementedError("This must be overwritten by the subclass.")
@classmethod
def _assign(cls, container, index, value):
r"""Assign an element in the container to the specified value.
Args:
container (obj): Object that element will be assigned to.
index (obj): Index in the container object where element will be
assigned.
value (obj): Value that will be assigned to the element in the
container object.
"""
raise NotImplementedError("This must be overwritten by the subclass.")
@classmethod
def _has_element(cls, container, index):
r"""Check to see if an index is in the container.
Args:
container (obj): Object that should be checked for index.
index (obj): Index that should be checked for.
Returns:
bool: True if the index is in the container.
"""
raise NotImplementedError("This must be overwritten by the subclass.")
@classmethod
def _get_element(cls, container, index, default):
r"""Get an element from the container if it exists, otherwise return
the default.
Args:
container (obj): Object that should be returned from.
index (obj): Index of element that should be returned.
default (obj): Default that should be returned if the index is not
in the container.
Returns:
object: Container contents at specified element.
"""
out = default
if cls._has_element(container, index):
out = container[index]
return out
@classmethod
def _encode_data_alias(cls, obj, typedef, func_encode, container_type=None):
r"""Encode an object's data using a sepcified function.
Args:
obj (object): Object to encode.
typedef (dict): Type definition that should be used to encode the
object.
func_encode (callable): Function that should be used to encode
elements in the container. Defaults to encode_data.
container_type (type, optional): Type that should be used for the
output container. Defaults to cls._container_type.
Returns:
string: Encoded object.
"""
if container_type is None:
container = cls._container_type()
else:
container = container_type()
vtypedef_avail = False
if isinstance(typedef, dict) and (cls._json_property in typedef):
vtypedef_avail = typedef[cls._json_property]
for k, v in cls._iterate(obj):
vtypedef = None
if vtypedef_avail:
vtypedef = cls._get_element(vtypedef_avail, k, None)
vbytes = func_encode(v, typedef=vtypedef)
cls._assign(container, k, vbytes)
return container
[docs] @classmethod
def encode_data(cls, obj, typedef, **kwargs):
r"""Encode an object's data.
Args:
obj (object): Object to encode.
typedef (dict): Type definition that should be used to encode the
object.
**kwargs: Additional keyword arguments are passed to the class
method _encode_data_alias.
Returns:
string: Encoded object.
"""
return cls._encode_data_alias(obj, typedef, encode_data, **kwargs)
[docs] @classmethod
def encode_data_readable(cls, obj, typedef, **kwargs):
r"""Encode an object's data in a readable format.
Args:
obj (object): Object to encode.
typedef (dict): Type definition that should be used to encode the
object.
**kwargs: Additional keyword arguments are passed to the class
method _encode_data_alias.
Returns:
string: Encoded object.
"""
return cls._encode_data_alias(obj, typedef, encode_data_readable, **kwargs)
[docs] @classmethod
def decode_data(cls, obj, typedef):
r"""Decode an object.
Args:
obj (string): Encoded object to decode.
typedef (dict): Type definition that should be used to decode the
object.
Returns:
object: Decoded object.
"""
container = cls._container_type()
if cls._json_property in typedef:
for k, v in cls._iterate(obj):
vtypedef = cls._get_element(typedef[cls._json_property], k, {})
vcls = get_type_class(vtypedef['type'])
cls._assign(container, k, vcls.decode_data(v, vtypedef))
return container
[docs] @classmethod
def coerce_type(cls, obj, typedef=None, **kwargs):
r"""Coerce objects of specific types to match the data type.
Args:
obj (object): Object to be coerced.
typedef (dict, optional): Type defintion that object should be
coerced to. Defaults to None.
**kwargs: Additional keyword arguments are metadata entries that may
aid in coercing the type.
Returns:
object: Coerced object.
Raises:
RuntimeError: If obj is a dictionary, but key_order is not provided.
"""
if not (isinstance(typedef, dict)
and isinstance(typedef.get(cls._json_property, None),
cls.python_types)
and isinstance(obj, cls.python_types)):
return obj
map_typedef = typedef[cls._json_property]
map_out = cls._container_type()
for k, v in cls._iterate(obj):
if cls._has_element(map_typedef, k):
cls._assign(map_out, k,
get_type_class(map_typedef[k]['type']).coerce_type(
v, typedef=map_typedef[k]))
else:
cls._assign(map_out, k, v)
return map_out
[docs] @classmethod
def transform_type(cls, obj, typedef=None):
r"""Transform an object based on type info.
Args:
obj (object): Object to transform.
typedef (dict): Type definition that should be used to transform the
object.
Returns:
object: Transformed object.
"""
if not (isinstance(typedef, dict)
and isinstance(typedef.get(cls._json_property, None),
cls.python_types)
and isinstance(obj, cls.python_types)):
return obj
map_typedef = typedef[cls._json_property]
map_out = cls._container_type()
for k, v in cls._iterate(obj):
if cls._has_element(map_typedef, k):
cls._assign(map_out, k,
get_type_class(map_typedef[k]['type']).transform_type(
v, typedef=map_typedef[k]))
else:
cls._assign(map_out, k, v)
return map_out
[docs] @classmethod
def extract_typedef(cls, metadata):
r"""Extract the minimum typedef required for this type from the provided
metadata.
Args:
metadata (dict): Message metadata.
Returns:
dict: Encoded type definition with unncessary properties removed.
"""
out = super(ContainerMetaschemaType, cls).extract_typedef(metadata)
if cls._json_property in out:
contents = out[cls._json_property]
if isinstance(contents, cls.python_types):
for k, v in cls._iterate(contents):
if 'type' in v:
vcls = get_type_class(v['type'])
cls._assign(contents, k, vcls.extract_typedef(v))
out[cls._json_property] = contents
return out
[docs] def update_typedef(self, **kwargs):
r"""Update the current typedef with new values.
Args:
**kwargs: All keyword arguments are considered to be new type
definitions. If they are a valid definition property, they
will be copied to the typedef associated with the instance.
Returns:
dict: A dictionary of keyword arguments that were not added to the
type definition.
"""
map = kwargs.get(self._json_property, None)
map_out = self._container_type()
if isinstance(map, self.python_types):
for k, v in self._iterate(map):
v_typedef = complete_typedef(v)
if self._has_element(self._typecls, k):
self._assign(map_out, k,
self._typecls[k].update_typedef(**v_typedef))
else:
self._assign(self._typecls, k,
get_type_class(v_typedef['type'])(**v_typedef))
self._assign(map, k, self._typecls[k]._typedef)
kwargs[self._json_property] = map
out = super(ContainerMetaschemaType, self).update_typedef(**kwargs)
if map_out:
out[self._json_property] = map_out
return out
@classmethod
def _generate_data(cls, typedef, **kwargs):
r"""Generate mock data for the specified type.
Args:
typedef (dict): Type definition.
Returns:
object: Python object of the specified type.
"""
out = cls._container_type()
for k, v in cls._iterate(typedef[cls._json_property]):
vcls = get_type_class(v['type'])
cls._assign(out, k, vcls.generate_data(v, **kwargs))
return out