Source code for yggdrasil.metaschema.datatypes.MetaschemaType

import six
import copy
import uuid
import jsonschema
from yggdrasil import constants
from yggdrasil.metaschema import (get_metaschema, get_validator, encoder,
                                  validate_instance, MetaschemaTypeError)
from yggdrasil.metaschema.datatypes import (
    MetaschemaTypeMeta, compare_schema,
    get_type_class, conversions, is_default_typedef)
from yggdrasil.metaschema.properties import get_metaschema_property


def _get_single_array_element(arr):
    return arr[0]


[docs]@six.add_metaclass(MetaschemaTypeMeta) class MetaschemaType(object): r"""Base type that should be subclassed by user defined types. Attributes should be overwritten to match the type. Arguments: **kwargs: All keyword arguments are assumed to be type definition properties which will be used to validate serialized/deserialized messages. Class Attributes: name (str): Name of the type for use in YAML files & form options. [REQUIRED] description (str): A short description of the type. [REQUIRED] properties (list): List of JSON schema properties that this type uses. These properties will be added to those form the paretn type class unless inherit_properties is False. Defaults to ['type', 'title']. definition_properties (list): Type properties that are required for YAML or form entries specifying the type. These will also be used to validate type definitions. These properties will be added to those from the parent type class unless inherit_properties is False. Defaults to ['type']. metadata_properties (list): Type properties that are required for deserializing instances of the type that have been serialized. These properties will be added to those from the parent class unless inherit_properties is False. Defaults to ['type']. extract_properties (list): Properties that will be extracted from the metadata of a message to construct the type definition. These properties will be added to those from the parent class unless inherit_properties is False. Defaults to ['type', 'title']. python_types (list): List of python types that this type encompasses. [REQUIRED]. specificity (int): Specificity of the type. Types with larger values are more specific while types with smaller values are more general. Base types have a specificity of 0. More specific types are checked first before more general ones. is_fixed (bool): True if the type is a fixed version of another type. See FixedMetaschemaType for details. inherit_properties (bool, list): If True, a type class's properties will be a combination of the parent class's properties and those explicitly defined in the child class. If False, a type class's properties will only be those explicitly defined in the child class. If a list of property attributes, only properties specified by those attributes will be inherited from the parent class. schema_file (str, None): If set, the class's schema is loaded from the specified file at registration and moved to the loaded_schema_file class attribute. Defaults to None. loaded_schema_file (str, None): The path to the file the schema for the type was loaded from if it was loaded. cross_language_support (bool): If True, this indicates the types should be serializable across most (if not all) of the supported languages. If False, this type is not required to be serializable except to/from Python. Defaults to True. """ name = 'base' description = 'A generic base type for users to build on.' properties = ['type', 'title'] definition_properties = ['type'] metadata_properties = ['type'] extract_properties = ['type', 'title'] python_types = [] specificity = 0 is_fixed = False inherit_properties = True schema_file = None loaded_schema_file = None cross_language_support = True _dont_register = False _empty_msg = b'' _replaces_existing = False _json_type = None def __init__(self, **typedef): self._typedef = {} typedef.setdefault('type', self.name) self.update_typedef(**typedef) # Methods to be overridden by subclasses
[docs] @classmethod def encode_data(cls, obj, typedef): r"""Encode an object's data. Args: obj (object): Object to encode. typedef (dict): Type definition that should be used to encode the object. Returns: string: Encoded object. """ raise NotImplementedError("Method must be overridden by the subclass.")
[docs] @classmethod def encode_data_readable(cls, obj, typedef): 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. Returns: string: Encoded object. """ return cls.encode_data(obj, typedef)
[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. """ raise NotImplementedError("Method must be overridden by the subclass.")
[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. """ return obj
[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. """ return obj
# Methods not to be modified by subclasses
[docs] @classmethod def issubtype(cls, t): r"""Determine if this type is a subclass of the provided type. Args: t (str, list): Type name or list of type names to check against. Returns: bool: True if this type is a subtype of the specified type t. """ if isinstance(t, list): return (cls.name in t) return (cls.name == t)
[docs] @classmethod def jsonschema_type_checker(cls, checker, instance): r"""Type checker for use with jsonschema >= 3.0.0. Args: checker (jsonschema.TypeChecker): Type checker class. instance (object): Object being checked. Returns: bool: True if instance could be of this type, False otherwise. """ return cls.validate(instance)
[docs] @classmethod def validate(cls, obj, raise_errors=False): r"""Validate an object to check if it could be of this type. Args: obj (object): Object to validate. raise_errors (bool, optional): If True, errors will be raised when the object fails to be validated. Defaults to False. Returns: bool: True if the object could be of this type, False otherwise. """ if not cls.python_types: raise NotImplementedError("Attribute 'python_types' must be set.") out = isinstance(obj, cls.python_types) if (not out) and raise_errors: raise ValueError(("Object of type '%s' is not one of the accepted " + "Python types (%s) for this type (%s).") % (type(obj), cls.python_types, cls.name)) return out
[docs] @classmethod def normalize(cls, obj): r"""Normalize an object, if possible, to conform to this type. Args: obj (object): Object to normalize. Returns: object: Normalized object. """ return obj
[docs] @classmethod def encode_type(cls, obj, typedef=None, is_validated=False, **kwargs): r"""Encode an object's type definition. Args: obj (object): Object to encode. typedef (dict, optional): Type properties that should be used to initialize the encoded type definition in certain cases. Defaults to None and is ignored. **kwargs: Additional keyword arguments are treated as additional schema properties. Raises: MetaschemaTypeError: If the object is not the correct type. Returns: dict: Encoded type definition. """ obj = cls.coerce_type(obj, typedef=typedef) if typedef is None: typedef = {} if not is_validated: if not cls.validate(obj): raise MetaschemaTypeError(f"Object could not be encoded as " f"'{cls.name}' type.") out = copy.deepcopy(kwargs) for x in cls.properties: itypedef = typedef.get(x, out.get(x, None)) if x == 'type': out['type'] = cls.name elif x == 'title': if itypedef is not None: out[x] = itypedef else: prop_cls = get_metaschema_property(x) out[x] = prop_cls.encode(obj, typedef=itypedef) return out
[docs] @classmethod def get_extract_properties(cls, metadata): r"""Get the list of properties that should be kept when extracting a typedef from message metadata. Args: metadata (dict): Metadata that typedef is being extracted from. Returns: list: Keywords that should be kept in the typedef. """ return copy.deepcopy(cls.extract_properties)
[docs] @classmethod def extract_typedef(cls, metadata, reqkeys=None): r"""Extract the minimum typedef required for this type from the provided metadata. Args: metadata (dict): Message metadata. reqkeys (list, optional): Set of keys to keep in the definition. Defaults to the required definition keys. Returns: dict: Encoded type definition with unncessary properties removed. """ out = copy.deepcopy(metadata) if reqkeys is None: reqkeys = cls.get_extract_properties(metadata) keylist = [k for k in out.keys()] for k in keylist: if k not in reqkeys: del out[k] 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. Raises: MetaschemaTypeError: If the current type does not match the type being updated to. """ typename0 = self._typedef.get('type', None) typename1 = kwargs.get('type', None) # Check typename to make sure this is possible if typename1 and typename0 and (typename1 != typename0): raise MetaschemaTypeError( "Cannot update typedef for type '%s' to be '%s'." % (typename0, typename1)) # pragma: debug # Copy over valid properties all_keys = [k for k in kwargs.keys()] # req_keys = self.definition_schema().get('required', []) for k in all_keys: # if k in req_keys: self._typedef[k] = kwargs.pop(k) # Validate self.validate_definition(self._typedef) return kwargs
[docs] @classmethod def metaschema(cls): r"""JSON meta schema for validating schemas for this type.""" return get_metaschema()
[docs] @classmethod def validator(cls): r"""JSON schema validator for the meta schema that includes added types.""" return get_validator()
[docs] @classmethod def definition_schema(cls): r"""JSON schema for validating a type definition schema.""" out = {'title': cls.name, 'description': cls.description, 'type': 'object', 'required': copy.deepcopy(cls.definition_properties), 'properties': {'type': {'enum': [cls.name]}}} return out
[docs] @classmethod def metadata_schema(cls): r"""JSON schema for validating a JSON serialization of the type.""" out = {'title': cls.name, 'description': cls.description, 'type': 'object', 'required': copy.deepcopy(cls.metadata_properties), 'properties': {'type': {'enum': [cls.name]}}} return out
[docs] @classmethod def validate_metadata(cls, obj, **kwargs): r"""Validates an encoded object. Args: obj (string): Encoded object to validate. **kwargs: Additional keyword arguments are passed to the validator. """ if ((isinstance(obj, dict) and ('type' in obj) and (obj['type'] != cls.name))): type_cls = get_type_class(obj['type']) if type_cls.is_fixed and type_cls.issubtype(cls.name): obj = type_cls.typedef_fixed2base(obj) # jsonschema.validate(obj, cls.metaschema(), cls=cls.validator()) # jsonschema.validate(obj, cls.metadata_schema(), cls=cls.validator()) return validate_instance(obj, cls.metadata_schema(), **kwargs)
[docs] @classmethod def validate_definition(cls, obj, **kwargs): r"""Validates a type definition. Args: obj (object): Type definition to validate. **kwargs: Additional keyword arguments are passed to the validator. """ # jsonschema.validate(obj, cls.metaschema(), cls=cls.validator()) # jsonschema.validate(obj, cls.definition_schema(), cls=cls.validator()) return validate_instance(obj, cls.definition_schema(), **kwargs)
[docs] @classmethod def validate_instance(cls, obj, typedef, **kwargs): r"""Validates an object against a type definition. Args: obj (object): Object to validate against a type definition. typedef (dict): Type definition to validate against. **kwargs: Additional keyword arguments are passed to the validator. """ # cls.validate_definition(typedef) # jsonschema.validate(obj, typedef, cls=cls.validator()) return validate_instance(obj, typedef, **kwargs)
[docs] @classmethod def normalize_definition(cls, obj): r"""Normalizes a type definition. Args: obj (object): Type definition to normalize. Returns: object: Normalized type definition. """ for x in cls.properties: if x not in obj: prop_cls = get_metaschema_property(x) obj = prop_cls.normalize_in_schema(obj) return obj
[docs] @classmethod def check_encoded(cls, metadata, typedef=None, raise_errors=False, typedef_validated=False, metadata_validated=False): r"""Checks if the metadata for an encoded object matches the type definition. Args: metadata (dict): Meta data to be tested. typedef (dict, optional): Type properties that object should be tested against. Defaults to None and object may have any values for the type properties (so long as they match the schema. raise_errors (bool, optional): If True, any errors determining that encoded object is not of this type will be raised. Defaults to False. typedef_validated (bool, optional): If True, the type definition is taken as already having been validated and will not be validated again during the encoding process. Defaults to False. metadata_validated (bool, optional): If True, the metadata definition is taken as already having been valdiated and will not be validated again. Defaults to False. Returns: bool: True if the metadata matches the type definition, False otherwise. """ if not metadata_validated: try: cls.validate_metadata(metadata) except jsonschema.exceptions.ValidationError: if raise_errors: raise return False if typedef is not None: if not typedef_validated: try: cls.validate_definition(typedef) except jsonschema.exceptions.ValidationError: if raise_errors: raise return False errors = [e for e in compare_schema(metadata, typedef)] if errors: error_msg = "Error(s) in comparison:" for e in errors: error_msg += ('\t%s' % e) if raise_errors: raise ValueError(error_msg) return False return True
[docs] @classmethod def check_decoded(cls, obj, typedef=None, raise_errors=False, typedef_validated=False): r"""Checks if an object is of the this type. Args: obj (object): Object to be tested. typedef (dict, optional): Type properties that object should be tested against. Defaults to None and is not used. raise_errors (bool, optional): If True, any errors determining that decoded object is not of this type will be raised. Defaults to False. typedef_validated (bool, optional): If True, the type definition is taken as already having been validated and will not be validated again during the encoding process. Defaults to False. Returns: bool: Truth of if the input object is of this type. """ if not cls.validate(obj, raise_errors=raise_errors): return False if typedef is None: return True # Validate definition if not typedef_validated: try: cls.validate_definition(typedef) except jsonschema.exceptions.ValidationError: if raise_errors: raise return False # Validate instance against definition try: cls.validate_instance(obj, typedef) except jsonschema.exceptions.ValidationError: if raise_errors: raise return False return True
[docs] @classmethod def encode(cls, obj, typedef=None, typedef_validated=False, dont_check=False, **kwargs): r"""Encode an object. Args: obj (object): Object to encode. typedef (dict, optional): Type properties that object should be tested against. Defaults to None and object may have any values for the type properties (so long as they match the schema. typedef_validated (bool, optional): If True, the type definition is taken as already having been validated and will not be validated again during the encoding process. Defaults to False. dont_check (bool, optional): If True, the object will not be checked against the type definition. Defaults to False. **kwargs: Additional keyword arguments are added to the metadata. Returns: tuple(dict, bytes): Encoded object with type definition and data serialized to bytes. Raises: ValueError: If the object does not match the type definition. ValueError: If the encoded metadata does not match the type definition. TypeError: If the encoded data is not of bytes type. """ # Coerce, then check object, then transform obj = cls.coerce_type(obj, typedef=typedef, typedef_validated=typedef_validated, **kwargs) if not dont_check: cls.check_decoded(obj, typedef, raise_errors=True, typedef_validated=typedef_validated) obj_t = cls.transform_type(obj, typedef) # Encode metadata = cls.encode_type(obj_t, typedef=typedef) data = cls.encode_data(obj_t, metadata) return metadata, data
[docs] @classmethod def decode(cls, metadata, data, typedef=None, typedef_validated=False, dont_check=False): r"""Decode an object. Args: metadata (dict): Meta data describing the data. data (bytes): Encoded data. typedef (dict, optional): Type properties that decoded object should be tested against. Defaults to None and object may have any values for the type properties (so long as they match the schema). typedef_validated (bool, optional): If True, the type definition is taken as already having been validated and will not be validated again during the encoding process. Defaults to False. dont_check (bool, optional): If True, the metadata will not be checked against the type definition. Defaults to False. Returns: object: Decoded object. Raises: ValueError: If the metadata does not match the type definition. ValueError: If the decoded object does not match type definition. """ conv_func = None if isinstance(metadata, dict): metatype = metadata.get('type', None) if (metatype not in [None, 'bytes']) and is_default_typedef(typedef): new_cls = get_type_class(metatype) return new_cls.decode(metadata, data, dont_check=dont_check) if metatype != cls.name: conv_func = conversions.get_conversion(metatype, cls.name) if (((conv_func is None) and (len(metadata.get('items', [])) == 1) and cls.check_encoded(metadata['items'][0], typedef))): conv_func = _get_single_array_element if (not conv_func) and (not dont_check): cls.check_encoded(metadata, typedef, raise_errors=True, typedef_validated=typedef_validated) if conv_func: new_cls = get_type_class(metadata['type']) out = conv_func(new_cls.decode(metadata, data, dont_check=dont_check)) else: out = cls.decode_data(data, metadata) out = cls.transform_type(out, typedef) return out
[docs] def serialize(self, obj, no_metadata=False, dont_encode=False, dont_check=False, max_header_size=0, **kwargs): r"""Serialize a message. Args: obj (object): Python object to be formatted. no_metadata (bool, optional): If True, no metadata will be added to the serialized message. Defaults to False. dont_encode (bool, optional): If True, the input message will not be encoded using type specific or JSON encoding. Defaults to False. dont_check (bool, optional): If True, the object being serialized will not be checked against the type definition. Defaults to False. max_header_size (int, optional): Maximum size that header should occupy in order to be sent in a single message. A value of 0 indicates that any size header is valid. Defaults to 0. **kwargs: Additional keyword arguments are added to the metadata. Returns: bytes, str: Serialized message. """ for k in ['size', 'data', 'datatype']: if k in kwargs: raise RuntimeError("'%s' is a reserved keyword in the metadata." % k) if ((isinstance(obj, bytes) and ((obj == constants.YGG_MSG_EOF) or kwargs.get('raw', False) or dont_encode))): metadata = kwargs data = obj is_raw = True else: typedef, data = self.encode(obj, typedef=self._typedef, typedef_validated=True, dont_check=dont_check, **kwargs) metadata = {'datatype': typedef} metadata.update(kwargs) is_raw = False if not is_raw: data = encoder.encode_json(data) if no_metadata: return data metadata['size'] = len(data) metadata.setdefault('id', str(uuid.uuid4())) header = (constants.YGG_MSG_HEAD + encoder.encode_json(metadata) + constants.YGG_MSG_HEAD) if (max_header_size > 0) and (len(header) > max_header_size): metadata_type = metadata metadata = {} for k in ['address', 'size', 'id', 'request_id', 'response_address', 'zmq_reply', 'zmq_reply_worker', 'model']: if k in metadata_type: metadata[k] = metadata_type.pop(k) assert(metadata) data = (encoder.encode_json(metadata_type) + constants.YGG_MSG_HEAD + data) metadata['size'] = len(data) metadata['type_in_data'] = True header = (constants.YGG_MSG_HEAD + encoder.encode_json(metadata) + constants.YGG_MSG_HEAD) if len(header) > max_header_size: # pragma: debug raise AssertionError(("The header is larger (%d) than the " "maximum (%d): %.100s...") % (len(header), max_header_size, header)) msg = header + data return msg
[docs] def deserialize(self, msg, no_data=False, metadata=None, dont_decode=False, dont_check=False): r"""Deserialize a message. Args: msg (str, bytes): Message to be deserialized. no_data (bool, optional): If True, only the metadata is returned. Defaults to False. metadata (dict, optional): Metadata that should be used to deserialize the message instead of the current header content. Defaults to None and is not used. dont_decode (bool, optional): If True, type specific and JSON decoding will not be used to decode the message. Defaults to False. dont_check (bool, optional): If True, the metadata will not be checked against the type definition. Defaults to False. Returns: tuple(obj, dict): Deserialized message and header information. Raises: TypeError: If msg is not bytes type (str on Python 2). ValueError: If msg does not contain the header separator. """ if not isinstance(msg, bytes): raise TypeError("Message to be deserialized is not bytes type.") # Check for header if msg.startswith(constants.YGG_MSG_HEAD): if metadata is not None: raise ValueError("Metadata in header and provided by keyword.") _, metadata, data = msg.split(constants.YGG_MSG_HEAD, 2) if len(metadata) == 0: metadata = dict(size=len(data)) else: metadata = encoder.decode_json(metadata) elif isinstance(metadata, dict) and metadata.get('type_in_data', False): assert(msg.count(constants.YGG_MSG_HEAD) == 1) typedef, data = msg.split(constants.YGG_MSG_HEAD, 1) if len(typedef) > 0: metadata.update(encoder.decode_json(typedef)) metadata.pop('type_in_data') metadata['size'] = len(data) else: data = msg if metadata is None: metadata = dict(size=len(msg)) if (((len(msg) > 0) and (msg != constants.YGG_MSG_EOF) and (not is_default_typedef(self._typedef)) and (not dont_decode))): raise ValueError("Header marker not in message.") # Set flags based on data metadata['incomplete'] = (len(data) < metadata['size']) if (data == constants.YGG_MSG_EOF): metadata['raw'] = True # Return based on flags if no_data: return metadata elif len(data) == 0: return self._empty_msg, metadata elif (metadata['incomplete'] or metadata.get('raw', False) or (metadata.get('type', None) == 'direct') or dont_decode): return data, metadata else: data = encoder.decode_json(data) obj = self.decode(metadata['datatype'], data, self._typedef, typedef_validated=True, dont_check=dont_check) return obj, metadata
# TESTING METHODS @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. """ raise NotImplementedError # pragma: debug
[docs] @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. """ cls.validate_definition(typedef) typedef = cls.normalize_definition(typedef) if hasattr(cls, 'example_data'): return cls.example_data return cls._generate_data(typedef, **kwargs)
[docs] @classmethod def get_test_data(cls, typedef=None): r"""object: Test data.""" if typedef is None: typedef = {'type': cls.name} return cls.generate_data(typedef)