""" Defines the Intermediate Representation that is generated by the frontend and fed to the backends. The goal of this module is to define all data types that are common to the languages and serialization formats we want to support. """ from __future__ import absolute_import, division, print_function, unicode_literals from abc import ABCMeta, abstractmethod from collections import OrderedDict, deque import copy import datetime import math import numbers import re import six from ..frontend.exception import InvalidSpec from ..frontend.ast import ( AstExampleField, AstExampleRef, AstTagRef, ) _MYPY = False if _MYPY: import typing # noqa: F401 # pylint: disable=import-error,unused-import,useless-suppression class ParameterError(Exception): """Raised when a data type is parameterized with a bad type or value.""" def generic_type_name(v): """ Return a descriptive type name that isn't Python specific. For example, an int type will return 'integer' rather than 'int'. """ if isinstance(v, AstExampleRef): return "reference" elif isinstance(v, numbers.Integral): # Must come before real numbers check since integrals are reals too return 'integer' elif isinstance(v, numbers.Real): return 'float' elif isinstance(v, (tuple, list)): return 'list' elif isinstance(v, six.string_types): return 'string' elif v is None: return 'null' else: return type(v).__name__ def record_custom_annotation_imports(annotation, namespace): """ Records imports for custom annotations in the given namespace. """ # first, check the annotation *type* if annotation.annotation_type.namespace.name != namespace.name: namespace.add_imported_namespace( annotation.annotation_type.namespace, imported_annotation_type=True) # second, check if we need to import the annotation itself # the annotation namespace is currently not actually used in the # backends, which reconstruct the annotation from the annotation # type directly. This could be changed in the future, and at # the IR level it makes sense to include the dependency if annotation.namespace.name != namespace.name: namespace.add_imported_namespace( annotation.namespace, imported_annotation=True) class DataType(object): """ Abstract class representing a data type. """ __metaclass__ = ABCMeta def __init__(self): """No-op. Exists so that introspection can be certain that an init method exists.""" @property def name(self): """Returns an easy to read name for the type.""" return self.__class__.__name__ @abstractmethod def check(self, val): """ Checks if a value specified in a spec (translated to a Python object) is a valid Python value for this type. Returns nothing, but can raise an exception. Args: val (object) Raises: ValueError """ @abstractmethod def check_example(self, ex_field): """ Checks if an example field from a spec is valid. Returns nothing, but can raise an exception. Args: ex_field (AstExampleField) Raises: InvalidSpec """ def __repr__(self): return self.name class Primitive(DataType): # pylint: disable=abstract-method def check_attr_repr(self, attr_field): try: self.check(attr_field.value) except ValueError as e: raise InvalidSpec(e.args[0], attr_field.lineno, attr_field.path) return attr_field.value class Composite(DataType): # pylint: disable=abstract-method """ Composite types are any data type which can be constructed using primitive data types and other composite types. """ def __init__(self): super(Composite, self).__init__() # contains custom annotations that apply to any containing data types (recursively) # format is (location, CustomAnnotation) to indicate a custom annotation is applied # to a location (Field or Alias) self.recursive_custom_annotations = None class Nullable(Composite): def __init__(self, data_type): super(Nullable, self).__init__() self.data_type = data_type def check(self, val): if val is not None: return self.data_type.check(val) def check_example(self, ex_field): if ex_field.value is not None: return self.data_type.check_example(ex_field) def check_attr_repr(self, attr_field): if attr_field.value is None: return None else: return self.data_type.check_attr_repr(attr_field) class Void(Primitive): def check(self, val): if val is not None: raise ValueError('void type can only be null') def check_example(self, ex_field): if ex_field.value is not None: raise InvalidSpec('example of void type must be null', ex_field.lineno, ex_field.path) def check_attr_repr(self, attr_field): raise NotImplementedError class Bytes(Primitive): def check(self, val): if not isinstance(val, (bytes, six.text_type)): raise ValueError('%r is not valid bytes' % val) def check_example(self, ex_field): if not isinstance(ex_field.value, (bytes, six.text_type)): raise InvalidSpec("'%s' is not valid bytes" % ex_field.value, ex_field.lineno, ex_field.path) def check_attr_repr(self, attr_field): try: self.check(attr_field.value) except ValueError as e: raise InvalidSpec(e.args[0], attr_field.lineno, attr_field.path) v = attr_field.value if isinstance(v, six.text_type): return v.encode('utf-8') else: return v class _BoundedInteger(Primitive): """ When extending, specify 'minimum' and 'maximum' as class variables. This is the range of values supported by the data type. """ # See minimum = None # type: typing.Optional[int] maximum = None # type: typing.Optional[int] def __init__(self, min_value=None, max_value=None): """ A more restrictive minimum or maximum value can be specified than the range inherent to the defined type. """ super(_BoundedInteger, self).__init__() if min_value is not None: if not isinstance(min_value, numbers.Integral): raise ParameterError('min_value must be an integral number') if min_value < self.minimum: raise ParameterError('min_value cannot be less than the ' 'minimum value for this type (%s < %s)' % (min_value, self.minimum)) if max_value is not None: if not isinstance(max_value, numbers.Integral): raise ParameterError('max_value must be an integral number') if max_value > self.maximum: raise ParameterError('max_value cannot be greater than the ' 'maximum value for this type (%s < %s)' % (max_value, self.maximum)) self.min_value = min_value self.max_value = max_value def check(self, val): if not isinstance(val, numbers.Integral): raise ValueError('%s is not a valid integer' % generic_type_name(val)) if not (self.minimum <= val <= self.maximum): raise ValueError('%d is not within range [%r, %r]' % (val, self.minimum, self.maximum)) if self.min_value is not None and val < self.min_value: raise ValueError('%d is less than %d' % (val, self.min_value)) if self.max_value is not None and val > self.max_value: raise ValueError('%d is greater than %d' % (val, self.max_value)) def check_example(self, ex_field): try: self.check(ex_field.value) except ValueError as e: raise InvalidSpec(e.args[0], ex_field.lineno, ex_field.path) def __repr__(self): return '%s()' % self.name class Int32(_BoundedInteger): minimum = -2**31 maximum = 2**31 - 1 class UInt32(_BoundedInteger): minimum = 0 maximum = 2**32 - 1 class Int64(_BoundedInteger): minimum = -2**63 maximum = 2**63 - 1 class UInt64(_BoundedInteger): minimum = 0 maximum = 2**64 - 1 class _BoundedFloat(Primitive): """ When extending, optionally specify 'minimum' and 'maximum' as class variables. This is the range of values supported by the data type. For a float64, there is no need to specify a minimum and maximum since Python's native float implementation is a float64/double. Therefore, any Python float will pass the data type range check automatically. """ # See minimum = None # type: typing.Optional[float] maximum = None # type: typing.Optional[float] def __init__(self, min_value=None, max_value=None): """ A more restrictive minimum or maximum value can be specified than the range inherent to the defined type. """ super(_BoundedFloat, self).__init__() if min_value is not None: if not isinstance(min_value, numbers.Real): raise ParameterError('min_value must be a real number') if not isinstance(min_value, float): try: min_value = float(min_value) except OverflowError: raise ParameterError('min_value is too small for a float') if self.minimum is not None and min_value < self.minimum: raise ParameterError( 'min_value cannot be less than the ' # pylint: disable=E1307 'minimum value for this type (%f < %f)' % (min_value, self.minimum) ) if max_value is not None: if not isinstance(max_value, numbers.Real): raise ParameterError('max_value must be a real number') if not isinstance(max_value, float): try: max_value = float(max_value) except OverflowError: raise ParameterError('max_value is too large for a float') if self.maximum is not None and max_value > self.maximum: raise ParameterError( 'max_value cannot be greater than the ' # pylint: disable=E1307 'maximum value for this type (%f < %f)' % (max_value, self.maximum) ) self.min_value = min_value self.max_value = max_value def check(self, val): if not isinstance(val, numbers.Real): raise ValueError('%s is not a valid real number' % generic_type_name(val)) if not isinstance(val, float): try: val = float(val) except OverflowError: raise ValueError('%r is too large for float' % val) if math.isnan(val) or math.isinf(val): # Parser doesn't support NaN or Inf yet. raise ValueError('%f values are not supported' % val) if self.minimum is not None and val < self.minimum: raise ValueError( '%f is less than %f' % # pylint: disable=E1307 (val, self.minimum) ) if self.maximum is not None and val > self.maximum: raise ValueError( '%f is greater than %f' % # pylint: disable=E1307 (val, self.maximum) ) if self.min_value is not None and val < self.min_value: raise ValueError('%f is less than %f' % (val, self.min_value)) if self.max_value is not None and val > self.max_value: raise ValueError('%f is greater than %f' % (val, self.min_value)) def check_example(self, ex_field): try: self.check(ex_field.value) except ValueError as e: raise InvalidSpec(e.args[0], ex_field.lineno, ex_field.path) def __repr__(self): return '%s()' % self.name class Float32(_BoundedFloat): # Maximum and minimums from the IEEE 754-1985 standard minimum = -3.40282 * 10**38 maximum = 3.40282 * 10**38 class Float64(_BoundedFloat): pass class Boolean(Primitive): def check(self, val): if not isinstance(val, bool): raise ValueError('%r is not a valid boolean' % val) def check_example(self, ex_field): try: self.check(ex_field.value) except ValueError as e: raise InvalidSpec(e.args[0], ex_field.lineno, ex_field.path) class String(Primitive): def __init__(self, min_length=None, max_length=None, pattern=None): super(String, self).__init__() if min_length is not None: if not isinstance(min_length, numbers.Integral): raise ParameterError('min_length must be an integral number') if min_length < 0: raise ParameterError('min_length must be >= 0') if max_length is not None: if not isinstance(max_length, numbers.Integral): raise ParameterError('max_length must be an integral number') if max_length < 1: raise ParameterError('max_length must be > 0') if min_length and max_length: if max_length < min_length: raise ParameterError('max_length must be >= min_length') self.min_length = min_length self.max_length = max_length self.pattern = pattern self.pattern_re = None if pattern: if not isinstance(pattern, six.string_types): raise ParameterError('pattern must be a string') try: self.pattern_re = re.compile(pattern) except re.error as e: raise ParameterError( 'could not compile regex pattern {!r}: {}'.format( pattern, e.args[0])) def check(self, val): if not isinstance(val, six.string_types): raise ValueError('%s is not a valid string' % generic_type_name(val)) elif self.max_length is not None and len(val) > self.max_length: raise ValueError("'%s' has more than %d character(s)" % (val, self.max_length)) elif self.min_length is not None and len(val) < self.min_length: raise ValueError("'%s' has fewer than %d character(s)" % (val, self.min_length)) elif self.pattern and not self.pattern_re.match(val): raise ValueError("'%s' did not match pattern '%s'" % (val, self.pattern)) def check_example(self, ex_field): try: self.check(ex_field.value) except ValueError as e: raise InvalidSpec(e.args[0], ex_field.lineno, ex_field.path) class Timestamp(Primitive): def __init__(self, fmt): super(Timestamp, self).__init__() if not isinstance(fmt, six.string_types): raise ParameterError('format must be a string') self.format = fmt def check(self, val): if not isinstance(val, six.string_types): raise ValueError('timestamp must be specified as a string') # Raises a ValueError if val is the incorrect format datetime.datetime.strptime(val, self.format) def check_example(self, ex_field): try: self.check(ex_field.value) except ValueError as e: raise InvalidSpec(e.args[0], ex_field.lineno, ex_field.path) def check_attr_repr(self, attr_field): try: self.check(attr_field.value) except ValueError as e: msg = e.args[0] if isinstance(msg, six.binary_type): # For Python 2 compatibility. msg = msg.decode('utf-8') raise InvalidSpec(msg, attr_field.lineno, attr_field.path) return datetime.datetime.strptime(attr_field.value, self.format) class List(Composite): def __init__(self, data_type, min_items=None, max_items=None): super(List, self).__init__() self.data_type = data_type if min_items is not None and min_items < 0: raise ParameterError('min_items must be >= 0') if max_items is not None and max_items < 1: raise ParameterError('max_items must be > 0') if min_items and max_items and max_items < min_items: raise ParameterError('max_length must be >= min_length') self.min_items = min_items self.max_items = max_items def check(self, val): raise NotImplementedError def check_example(self, ex_field): try: self._check_list_container(ex_field.value) for item in ex_field.value: new_ex_field = AstExampleField( ex_field.path, ex_field.lineno, ex_field.lexpos, ex_field.name, item) self.data_type.check_example(new_ex_field) except ValueError as e: raise InvalidSpec(e.args[0], ex_field.lineno, ex_field.path) def _check_list_container(self, val): if not isinstance(val, list): raise ValueError('%s is not a valid list' % generic_type_name(val)) elif self.max_items is not None and len(val) > self.max_items: raise ValueError('list has more than %s item(s)' % self.max_items) elif self.min_items is not None and len(val) < self.min_items: raise ValueError('list has fewer than %s item(s)' % self.min_items) class Map(Composite): def __init__(self, key_data_type, value_data_type): super(Map, self).__init__() if not isinstance(key_data_type, String): raise ParameterError("Only String primitives are supported as key types.") self.key_data_type = key_data_type self.value_data_type = value_data_type def check(self, val): raise NotImplementedError def check_example(self, ex_field): if not isinstance(ex_field.value, dict): raise ValueError("%s is not a valid map" % generic_type_name(ex_field.value)) for k, v in ex_field.value.items(): ex_key_field = self._make_ex_field(ex_field, k) ex_value_field = self._make_ex_field(ex_field, v) self.key_data_type.check_example(ex_key_field) self.value_data_type.check_example(ex_value_field) def _make_ex_field(self, ex_field, value): return AstExampleField( ex_field.path, ex_field.lineno, ex_field.lexpos, ex_field.name, value) def doc_unwrap(raw_doc): """ Applies two transformations to raw_doc: 1. N consecutive newlines are converted into N-1 newlines. 2. A lone newline is converted to a space, which basically unwraps text. Returns a new string, or None if the input was None. """ if raw_doc is None: return None docstring = '' consecutive_newlines = 0 # Remove all leading and trailing whitespace in the documentation block for c in raw_doc.strip(): if c == '\n': consecutive_newlines += 1 if consecutive_newlines > 1: docstring += c else: if consecutive_newlines == 1: docstring += ' ' consecutive_newlines = 0 docstring += c return docstring class Field(object): """ Represents a field in a composite type. """ def __init__(self, name, data_type, doc, ast_node): """ Creates a new Field. :param str name: Name of the field. :param Type data_type: The type of variable for of this field. :param str doc: Documentation for the field. :param ast_node: Raw field definition from the parser. :type ast_node: stone.frontend.ast.AstField """ self.name = name self.data_type = data_type self.raw_doc = doc self.doc = doc_unwrap(doc) self._ast_node = ast_node self.redactor = None self.omitted_caller = None self.deprecated = None self.preview = None self.custom_annotations = [] def set_annotations(self, annotations): if not annotations: return for annotation in annotations: if isinstance(annotation, Deprecated): if self.deprecated: raise InvalidSpec("Deprecated value already set as %r." % str(self.deprecated), self._ast_node.lineno) if self.preview: raise InvalidSpec("'Deprecated' and 'Preview' can\'t both be set.", self._ast_node.lineno) self.deprecated = True self.doc = 'Field is deprecated. {}'.format(self.doc) elif isinstance(annotation, Omitted): if self.omitted_caller: raise InvalidSpec("Omitted caller already set as %r." % str(self.omitted_caller), self._ast_node.lineno) self.omitted_caller = annotation.omitted_caller self.doc = 'Field is only returned for "{}" callers. {}'.format( str(self.omitted_caller), self.doc) elif isinstance(annotation, Preview): if self.preview: raise InvalidSpec("Preview value already set as %r." % str(self.preview), self._ast_node.lineno) if self.deprecated: raise InvalidSpec("'Deprecated' and 'Preview' can\'t both be set.", self._ast_node.lineno) self.preview = True self.doc = 'Field is in preview mode - do not rely on in production. {}'.format( self.doc ) elif isinstance(annotation, Redacted): # Make sure we don't set multiple conflicting annotations on one field if self.redactor: raise InvalidSpec("Redactor already set as %r." % str(self.redactor), self._ast_node.lineno) self.redactor = annotation elif isinstance(annotation, CustomAnnotation): self.custom_annotations.append(annotation) else: raise InvalidSpec( 'Annotation %r not recognized for field.' % annotation, self._ast_node.lineno) def __repr__(self): return 'Field(%r, %r)' % (self.name, self.data_type) class StructField(Field): """ Represents a field of a struct. """ def __init__(self, name, data_type, doc, ast_node): """ Creates a new Field. :param str name: Name of the field. :param Type data_type: The type of variable for of this field. :param str doc: Documentation for the field. :param ast_node: Raw field definition from the parser. :type ast_node: stone.frontend.ast.AstField """ super(StructField, self).__init__(name, data_type, doc, ast_node) self.has_default = False self._default = None def set_default(self, default): self.has_default = True self._default = default @property def default(self): if not self.has_default: raise Exception('Type has no default') else: return self._default def check_attr_repr(self, attr): if attr is not None: attr = self.data_type.check_attr_repr(attr) if attr is None: if self.has_default: return self.default _, unwrapped_nullable, _ = unwrap(self.data_type) if unwrapped_nullable: return None else: raise KeyError(self.name) return attr def __repr__(self): return 'StructField(%r, %r, %r)' % (self.name, self.data_type, self.omitted_caller) class UnionField(Field): """ Represents a field of a union. """ def __init__(self, name, data_type, doc, ast_node, catch_all=False): super(UnionField, self).__init__(name, data_type, doc, ast_node) self.catch_all = catch_all def __repr__(self): return 'UnionField(%r, %r, %r, %r)' % (self.name, self.data_type, self.catch_all, self.omitted_caller) class UserDefined(Composite): """ These are types that are defined directly in specs. """ DEFAULT_EXAMPLE_LABEL = 'default' def __init__(self, name, namespace, ast_node): """ When this is instantiated, the type is treated as a forward reference. Only when :meth:`set_attributes` is called is the type considered to be fully defined. :param str name: Name of type. :param stone.ir.Namespace namespace: The namespace this type is defined in. :param ast_node: Raw type definition from the parser. :type ast_node: stone.frontend.ast.AstTypeDef """ super(UserDefined, self).__init__() self._name = name self.namespace = namespace self._ast_node = ast_node self._is_forward_ref = True self.raw_doc = None self.doc = None self.fields = None self.parent_type = None self._raw_examples = None self._examples = None self._fields_by_name = None def set_attributes(self, doc, fields, parent_type=None): """ Fields are specified as a list so that order is preserved for display purposes only. (Might be used for certain serialization formats...) :param str doc: Description of type. :param list(Field) fields: Ordered list of fields for type. :param Optional[Composite] parent_type: The type this type inherits from. """ self.raw_doc = doc self.doc = doc_unwrap(doc) self.fields = fields self.parent_type = parent_type self._raw_examples = OrderedDict() self._examples = OrderedDict() self._fields_by_name = {} # Dict[str, Field] # Check that no two fields share the same name. for field in self.fields: if field.name in self._fields_by_name: orig_lineno = self._fields_by_name[field.name]._ast_node.lineno raise InvalidSpec("Field '%s' already defined on line %s." % (field.name, orig_lineno), field._ast_node.lineno) self._fields_by_name[field.name] = field # Check that the fields for this type do not match any of the fields of # its parents. cur_type = self.parent_type while cur_type: for field in self.fields: if field.name in cur_type._fields_by_name: lineno = cur_type._fields_by_name[field.name]._ast_node.lineno raise InvalidSpec( "Field '%s' already defined in parent '%s' on line %d." % (field.name, cur_type.name, lineno), field._ast_node.lineno) cur_type = cur_type.parent_type # Import namespaces containing any custom annotations # Note: we don't need to do this for builtin annotations because # they are treated as globals at the IR level for field in self.fields: for annotation in field.custom_annotations: record_custom_annotation_imports(annotation, self.namespace) # Indicate that the attributes of the type have been populated. self._is_forward_ref = False @property def all_fields(self): raise NotImplementedError def has_documented_type_or_fields(self, include_inherited_fields=False): """Returns whether this type, or any of its fields, are documented. Use this when deciding whether to create a block of documentation for this type. """ if self.doc: return True else: return self.has_documented_fields(include_inherited_fields) def has_documented_fields(self, include_inherited_fields=False): """Returns whether at least one field is documented.""" fields = self.all_fields if include_inherited_fields else self.fields for field in fields: if field.doc: return True return False def get_all_omitted_callers(self): """Returns all unique omitted callers for the object.""" return {f.omitted_caller for f in self.fields if f.omitted_caller} @property def name(self): return self._name def copy(self): return copy.deepcopy(self) def prepend_field(self, field): self.fields.insert(0, field) def get_examples(self, compact=False): """ Returns an OrderedDict mapping labels to Example objects. Args: compact (bool): If True, union members of void type are converted to their compact representation: no ".tag" key or containing dict, just the tag as a string. """ # Copy it just in case the caller wants to mutate the object. examples = copy.deepcopy(self._examples) if not compact: return examples def make_compact(d): # Traverse through dicts looking for ones that have a lone .tag # key, which can be converted into the compact form. if not isinstance(d, dict): return for key in d: if isinstance(d[key], dict): inner_d = d[key] if len(inner_d) == 1 and '.tag' in inner_d: d[key] = inner_d['.tag'] else: make_compact(inner_d) if isinstance(d[key], list): for item in d[key]: make_compact(item) for example in examples.values(): if (isinstance(example.value, dict) and len(example.value) == 1 and '.tag' in example.value): # Handle the case where the top-level of the example can be # made compact. example.value = example.value['.tag'] else: make_compact(example.value) return examples class Example(object): """An example of a struct or union type.""" def __init__(self, label, text, value, ast_node=None): assert isinstance(label, six.text_type), type(label) self.label = label assert isinstance(text, (six.text_type, type(None))), type(text) self.text = doc_unwrap(text) if text else text assert isinstance(value, (six.text_type, OrderedDict)), type(value) self.value = value self._ast_node = ast_node def __repr__(self): return 'Example({!r}, {!r}, {!r})'.format( self.label, self.text, self.value) class Struct(UserDefined): """ Defines a product type: Composed of other primitive and/or struct types. """ composite_type = 'struct' def set_attributes(self, doc, fields, parent_type=None): """ See :meth:`Composite.set_attributes` for parameter definitions. """ if parent_type: assert isinstance(parent_type, Struct) self.subtypes = [] # These are only set if this struct enumerates subtypes. self._enumerated_subtypes = None # Optional[List[Tuple[str, DataType]]] self._is_catch_all = None # Optional[Bool] super(Struct, self).set_attributes(doc, fields, parent_type) if self.parent_type: self.parent_type.subtypes.append(self) def check(self, val): raise NotImplementedError def check_example(self, ex_field): if not isinstance(ex_field.value, AstExampleRef): raise InvalidSpec( "example must reference label of '%s'" % self.name, ex_field.lineno, ex_field.path) def check_attr_repr(self, attrs): # Since we mutate it, let's make a copy to avoid mutating the argument. attrs = attrs.copy() validated_attrs = {} for field in self.all_fields: attr = field.check_attr_repr(attrs.pop(field.name, None)) validated_attrs[field.name] = attr if attrs: attr_name, attr_field = attrs.popitem() raise InvalidSpec( "Route attribute '%s' is not defined in 'stone_cfg.Route'." % attr_name, attr_field.lineno, attr_field.path) return validated_attrs @property def all_fields(self): """ Returns an iterator of all fields. Required fields before optional fields. Super type fields before type fields. """ return self.all_required_fields + self.all_optional_fields def _filter_fields(self, filter_function): """ Utility to iterate through all fields (super types first) of a type. :param filter: A function that takes in a Field object. If it returns True, the field is part of the generated output. If False, it is omitted. """ fields = [] if self.parent_type: fields.extend(self.parent_type._filter_fields(filter_function)) fields.extend(filter(filter_function, self.fields)) return fields @property def all_required_fields(self): """ Returns an iterator that traverses required fields in all super types first, and then for this type. """ def required_check(f): return not is_nullable_type(f.data_type) and not f.has_default return self._filter_fields(required_check) @property def all_optional_fields(self): """ Returns an iterator that traverses optional fields in all super types first, and then for this type. """ def optional_check(f): return is_nullable_type(f.data_type) or f.has_default return self._filter_fields(optional_check) def has_enumerated_subtypes(self): """ Whether this struct enumerates its subtypes. """ return bool(self._enumerated_subtypes) def get_enumerated_subtypes(self): """ Returns a list of subtype fields. Each field has a `name` attribute which is the tag for the subtype. Each field also has a `data_type` attribute that is a `Struct` object representing the subtype. """ assert self._enumerated_subtypes is not None return self._enumerated_subtypes def is_member_of_enumerated_subtypes_tree(self): """ Whether this struct enumerates subtypes or is a struct that is enumerated by its parent type. Because such structs are serialized and deserialized differently, use this method to detect these. """ return (self.has_enumerated_subtypes() or (self.parent_type and self.parent_type.has_enumerated_subtypes())) def is_catch_all(self): """ Indicates whether this struct should be used in the event that none of its known enumerated subtypes match a received type tag. Use this method only if the struct has enumerated subtypes. Returns: bool """ assert self._enumerated_subtypes is not None return self._is_catch_all def set_enumerated_subtypes(self, subtype_fields, is_catch_all): """ Sets the list of "enumerated subtypes" for this struct. This differs from regular subtyping in that each subtype is associated with a tag that is used in the serialized format to indicate the subtype. Also, this list of subtypes was explicitly defined in an "inner-union" in the specification. The list of fields must include all defined subtypes of this struct. NOTE(kelkabany): For this to work with upcoming forward references, the hierarchy of parent types for this struct must have had this method called on them already. :type subtype_fields: List[UnionField] """ assert self._enumerated_subtypes is None, \ 'Enumerated subtypes already set.' assert isinstance(is_catch_all, bool), type(is_catch_all) self._is_catch_all = is_catch_all self._enumerated_subtypes = [] if self.parent_type: raise InvalidSpec( "'%s' enumerates subtypes so it cannot extend another struct." % self.name, self._ast_node.lineno, self._ast_node.path) # Require that if this struct enumerates subtypes, its parent (and thus # the entire hierarchy above this struct) does as well. if self.parent_type and not self.parent_type.has_enumerated_subtypes(): raise InvalidSpec( "'%s' cannot enumerate subtypes if parent '%s' does not." % (self.name, self.parent_type.name), self._ast_node.lineno, self._ast_node.path) enumerated_subtype_names = set() # Set[str] for subtype_field in subtype_fields: path = subtype_field._ast_node.path lineno = subtype_field._ast_node.lineno # Require that a subtype only has a single type tag. if subtype_field.data_type.name in enumerated_subtype_names: raise InvalidSpec( "Subtype '%s' can only be specified once." % subtype_field.data_type.name, lineno, path) # Require that a subtype has this struct as its parent. if subtype_field.data_type.parent_type != self: raise InvalidSpec( "'%s' is not a subtype of '%s'." % (subtype_field.data_type.name, self.name), lineno, path) # Check for subtype tags that conflict with this struct's # non-inherited fields. if subtype_field.name in self._fields_by_name: # Since the union definition comes first, use its line number # as the source of the field's original declaration. orig_field = self._fields_by_name[subtype_field.name] raise InvalidSpec( "Field '%s' already defined on line %d." % (subtype_field.name, lineno), orig_field._ast_node.lineno, orig_field._ast_node.path) # Walk up parent tree hierarchy to ensure no field conflicts. # Checks for conflicts with subtype tags and regular fields. cur_type = self.parent_type while cur_type: if subtype_field.name in cur_type._fields_by_name: orig_field = cur_type._fields_by_name[subtype_field.name] raise InvalidSpec( "Field '%s' already defined in parent '%s' (%s:%d)." % (subtype_field.name, cur_type.name, orig_field._ast_node.path, orig_field._ast_node.lineno), lineno, path) cur_type = cur_type.parent_type # Note the discrepancy between `fields` which contains only the # struct fields, and `_fields_by_name` which contains the struct # fields and enumerated subtype fields. self._fields_by_name[subtype_field.name] = subtype_field enumerated_subtype_names.add(subtype_field.data_type.name) self._enumerated_subtypes.append(subtype_field) assert len(self._enumerated_subtypes) > 0 # Check that all known subtypes are listed in the enumeration. for subtype in self.subtypes: if subtype.name not in enumerated_subtype_names: raise InvalidSpec( "'%s' does not enumerate all subtypes, missing '%s'" % (self.name, subtype.name), self._ast_node.lineno) def get_all_subtypes_with_tags(self): """ Unlike other enumerated-subtypes-related functionality, this method returns not just direct subtypes, but all subtypes of this struct. The tag of each subtype is the list of tags from which the type descends. This method only applies to structs that enumerate subtypes. Use this when you need to generate a lookup table for a root struct that maps a generated class representing a subtype to the tag it needs in the serialized format. Returns: List[Tuple[List[String], Struct]] """ assert self.has_enumerated_subtypes(), 'Enumerated subtypes not set.' subtypes_with_tags = [] # List[Tuple[List[String], Struct]] fifo = deque([subtype_field.data_type for subtype_field in self.get_enumerated_subtypes()]) # Traverse down the hierarchy registering subtypes as they're found. while fifo: data_type = fifo.popleft() subtypes_with_tags.append((data_type._get_subtype_tags(), data_type)) if data_type.has_enumerated_subtypes(): for subtype_field in data_type.get_enumerated_subtypes(): fifo.append(subtype_field.data_type) return subtypes_with_tags def _get_subtype_tags(self): """ Returns a list of type tags that refer to this type starting from the base of the struct hierarchy. """ assert self.is_member_of_enumerated_subtypes_tree(), \ 'Not a part of a subtypes tree.' cur = self.parent_type cur_dt = self tags = [] while cur: assert cur.has_enumerated_subtypes() for subtype_field in cur.get_enumerated_subtypes(): if subtype_field.data_type is cur_dt: tags.append(subtype_field.name) break else: assert False, 'Could not find?!' cur_dt = cur cur = cur.parent_type tags.reverse() return tuple(tags) def _add_example(self, example): """Adds a "raw example" for this type. This does basic sanity checking to ensure that the example is valid (required fields specified, no unknown fields, correct types, ...). The example is not available via :meth:`get_examples` until :meth:`_compute_examples` is called. Args: example (stone.frontend.ast.AstExample): An example of this type. """ if self.has_enumerated_subtypes(): self._add_example_enumerated_subtypes_helper(example) else: self._add_example_helper(example) def _add_example_enumerated_subtypes_helper(self, example): """Validates examples for structs with enumerated subtypes.""" if len(example.fields) != 1: raise InvalidSpec( 'Example for struct with enumerated subtypes must only ' 'specify one subtype tag.', example.lineno, example.path) # Extract the only tag in the example. example_field = list(example.fields.values())[0] tag = example_field.name val = example_field.value if not isinstance(val, AstExampleRef): raise InvalidSpec( "Example of struct with enumerated subtypes must be a " "reference to a subtype's example.", example_field.lineno, example_field.path) for subtype_field in self.get_enumerated_subtypes(): if subtype_field.name == tag: self._raw_examples[example.label] = example break else: raise InvalidSpec( "Unknown subtype tag '%s' in example." % tag, example_field.lineno, example_field.path) def _add_example_helper(self, example): """Validates examples for structs without enumerated subtypes.""" # Check for fields in the example that don't belong. for label, example_field in example.fields.items(): if not any(label == f.name for f in self.all_fields): raise InvalidSpec( "Example for '%s' has unknown field '%s'." % (self.name, label), example_field.lineno, example_field.path, ) for field in self.all_fields: if field.name in example.fields: example_field = example.fields[field.name] try: field.data_type.check_example(example_field) except InvalidSpec as e: e.msg = "Bad example for field '{}': {}".format( field.name, e.msg) raise elif field.has_default or isinstance(field.data_type, Nullable): # These don't need examples. pass else: raise InvalidSpec( "Missing field '%s' in example." % field.name, example.lineno, example.path) self._raw_examples[example.label] = example def _has_example(self, label): """Whether this data type has an example with the given ``label``.""" return label in self._raw_examples def _compute_examples(self): """ Populates the ``_examples`` instance attribute by computing full examples for each label in ``_raw_examples``. The logic in this method is separate from :meth:`_add_example` because this method requires that every type have ``_raw_examples`` assigned for resolving example references. """ for label in self._raw_examples: self._examples[label] = self._compute_example(label) def _compute_example(self, label): if self.has_enumerated_subtypes(): return self._compute_example_enumerated_subtypes(label) else: return self._compute_example_flat_helper(label) def _compute_example_flat_helper(self, label): """ From the "raw example," resolves references to examples of other data types to compute the final example. Returns an Example object. The `value` attribute contains a JSON-serializable representation of the example. """ assert label in self._raw_examples, label example = self._raw_examples[label] def deref_example_ref(dt, val): dt, _ = unwrap_nullable(dt) if not dt._has_example(val.label): raise InvalidSpec( "Reference to example for '%s' with label '%s' " "does not exist." % (dt.name, val.label), val.lineno, val.path) return dt._compute_example(val.label).value # Do a deep copy of the example because we're going to mutate it. ex_val = OrderedDict() def get_json_val(dt, val): if isinstance(val, AstExampleRef): # Embed references to other examples directly. return deref_example_ref(dt, val) elif isinstance(val, TagRef): return val.union_data_type._compute_example(val.tag_name).value elif isinstance(val, list): dt, _ = unwrap_nullable(dt) return [get_json_val(dt.data_type, v) for v in val] elif isinstance(val, dict): dt, _ = unwrap_nullable(dt) if is_alias(dt): return val return {k: get_json_val(dt.value_data_type, v) for (k, v) in val.items()} else: return val for field in self.all_fields: if field.name in example.fields: example_field = example.fields[field.name] if example_field.value is None: # Serialized format doesn't include fields with null. pass else: ex_val[field.name] = get_json_val( field.data_type, example_field.value) elif field.has_default: ex_val[field.name] = get_json_val( field.data_type, field.default) return Example(example.label, example.text, ex_val, ast_node=example) def _compute_example_enumerated_subtypes(self, label): """ Analogous to :meth:`_compute_example_flat_helper` but for structs with enumerated subtypes. """ assert label in self._raw_examples, label example = self._raw_examples[label] example_field = list(example.fields.values())[0] for subtype_field in self.get_enumerated_subtypes(): if subtype_field.name == example_field.name: data_type = subtype_field.data_type break ref = example_field.value if not data_type._has_example(ref.label): raise InvalidSpec( "Reference to example for '%s' with label '%s' does not " "exist." % (data_type.name, ref.label), ref.lineno, ref.path) ordered_value = OrderedDict([('.tag', example_field.name)]) flat_example = data_type._compute_example_flat_helper(ref.label) ordered_value.update(flat_example.value) flat_example.value = ordered_value return flat_example def __repr__(self): return 'Struct(%r, %r)' % (self.name, self.fields) class Union(UserDefined): """Defines a tagged union. Fields are variants.""" composite_type = 'union' def __init__(self, name, namespace, ast_node, closed): super(Union, self).__init__(name, namespace, ast_node) self.closed = closed # TODO: Why is this a different signature than the parent? Is this # intentional? def set_attributes(self, doc, fields, # pylint: disable=arguments-differ parent_type=None, catch_all_field=None): """ :param UnionField catch_all_field: The field designated as the catch-all. This field should be a member of the list of fields. See :meth:`Composite.set_attributes` for parameter definitions. """ if parent_type: assert isinstance(parent_type, Union) super(Union, self).set_attributes(doc, fields, parent_type) self.catch_all_field = catch_all_field self.parent_type = parent_type def check(self, val): assert isinstance(val, TagRef) for field in self.all_fields: if val.tag_name == field.name: if not is_void_type(field.data_type): raise ValueError( "invalid reference to non-void option '%s'" % val.tag_name) break else: raise ValueError( "invalid reference to unknown tag '%s'" % val.tag_name) def check_example(self, ex_field): if not isinstance(ex_field.value, AstExampleRef): raise InvalidSpec( "example must reference label of '%s'" % self.name, ex_field.lineno, ex_field.path) def check_attr_repr(self, attr_field): if not isinstance(attr_field.value, AstTagRef): raise InvalidSpec( 'Expected union tag as value.', attr_field.lineno, attr_field.path) tag_ref = TagRef(self, attr_field.value.tag) try: self.check(tag_ref) except ValueError as e: raise InvalidSpec(e.args[0], attr_field.lineno, attr_field.path) return tag_ref @property def all_fields(self): """ Returns a list of all fields. Subtype fields come before this type's fields. """ fields = [] if self.parent_type: fields.extend(self.parent_type.all_fields) fields.extend([f for f in self.fields]) return fields def _add_example(self, example): """Adds a "raw example" for this type. This does basic sanity checking to ensure that the example is valid (required fields specified, no unknown fields, correct types, ...). The example is not available via :meth:`get_examples` until :meth:`_compute_examples` is called. Args: example (stone.frontend.ast.AstExample): An example of this type. """ if len(example.fields) != 1: raise InvalidSpec( 'Example for union must specify exactly one tag.', example.lineno, example.path) # Extract the only tag in the example. example_field = list(example.fields.values())[0] tag = example_field.name # Find the union member that corresponds to the tag. for field in self.all_fields: if tag == field.name: break else: # Error: Tag doesn't match any union member. raise InvalidSpec( "Unknown tag '%s' in example." % tag, example.lineno, example.path ) # TODO: are we always guaranteed at least one field? # pylint: disable=undefined-loop-variable try: field.data_type.check_example(example_field) except InvalidSpec as e: e.msg = "Bad example for field '{}': {}".format( field.name, e.msg) raise self._raw_examples[example.label] = example def _has_example(self, label): """Whether this data type has an example with the given ``label``.""" if label in self._raw_examples: return True else: for field in self.all_fields: dt, _ = unwrap_nullable(field.data_type) if not is_user_defined_type(dt) and not is_void_type(dt): continue if label == field.name: return True else: return False def _compute_examples(self): """ Populates the ``_examples`` instance attribute by computing full examples for each label in ``_raw_examples``. The logic in this method is separate from :meth:`_add_example` because this method requires that every type have ``_raw_examples`` assigned for resolving example references. """ for label in self._raw_examples: self._examples[label] = self._compute_example(label) # Add examples for each void union member. for field in self.all_fields: dt, _ = unwrap_nullable(field.data_type) if is_void_type(dt): self._examples[field.name] = \ Example( field.name, None, OrderedDict([('.tag', field.name)])) def _compute_example(self, label): """ From the "raw example," resolves references to examples of other data types to compute the final example. Returns an Example object. The `value` attribute contains a JSON-serializable representation of the example. """ if label in self._raw_examples: example = self._raw_examples[label] def deref_example_ref(dt, val): dt, _ = unwrap_nullable(dt) if not dt._has_example(val.label): raise InvalidSpec( "Reference to example for '%s' with label '%s' " "does not exist." % (dt.name, val.label), val.lineno, val.path) return dt._compute_example(val.label).value def get_json_val(dt, val): if isinstance(val, AstExampleRef): # Embed references to other examples directly. return deref_example_ref(dt, val) elif isinstance(val, list): return [get_json_val(dt.data_type, v) for v in val] else: return val example_field = list(example.fields.values())[0] # Do a deep copy of the example because we're going to mutate it. ex_val = OrderedDict([('.tag', example_field.name)]) for field in self.all_fields: if field.name == example_field.name: break # TODO: are we always guaranteed at least one field? # pylint: disable=undefined-loop-variable data_type, _ = unwrap_nullable(field.data_type) inner_ex_val = get_json_val(data_type, example_field.value) if (isinstance(data_type, Struct) and not data_type.has_enumerated_subtypes()): ex_val.update(inner_ex_val) else: if inner_ex_val is not None: ex_val[field.name] = inner_ex_val return Example(example.label, example.text, ex_val, ast_node=example) else: # Try to fallback to a union member with tag matching the label # with a data type that is composite or void. for field in self.all_fields: if label == field.name: break else: raise AssertionError('No example for label %r' % label) # TODO: are we always guaranteed at least one field? # pylint: disable=undefined-loop-variable assert is_void_type(field.data_type) return Example( field.name, field.doc, OrderedDict([('.tag', field.name)])) def unique_field_data_types(self): """ Checks if all variants have different data types. If so, the selected variant can be determined just by the data type of the value without needing a field name / tag. In some languages, this lets us make a shortcut """ data_type_names = set() for field in self.fields: if not is_void_type(field.data_type): if field.data_type.name in data_type_names: return False else: data_type_names.add(field.data_type.name) else: return True def __repr__(self): return 'Union(%r, %r)' % (self.name, self.fields) class TagRef(object): """ Used when an ID in Stone refers to a tag of a union. TODO(kelkabany): Support tag values. """ def __init__(self, union_data_type, tag_name): self.union_data_type = union_data_type self.tag_name = tag_name def __repr__(self): return 'TagRef(%r, %r)' % (self.union_data_type, self.tag_name) class AnnotationTypeParam(object): """ A parameter that can be supplied to a custom annotation type. """ def __init__(self, name, data_type, doc, has_default, default, ast_node): self.name = name self.data_type = data_type self.raw_doc = doc self.doc = doc_unwrap(doc) self.has_default = has_default self.default = default self._ast_node = ast_node if self.has_default: try: self.data_type.check(self.default) except ValueError as e: raise InvalidSpec('Default value for parameter %s is invalid: %s' % ( self.name, e), self._ast_node.lineno, self._ast_node.path) class AnnotationType(object): """ Used when a spec defines a custom annotation type. """ def __init__(self, name, namespace, doc, params): self.name = name self.namespace = namespace self.raw_doc = doc self.doc = doc_unwrap(doc) self.params = params self._params_by_name = {} # type: typing.Dict[str, AnnotationTypeParam] for param in self.params: if param.name in self._params_by_name: orig_lineno = self._params_by_name[param.name]._ast_node.lineno raise InvalidSpec("Parameter '%s' already defined on line %s." % (param.name, orig_lineno), param._ast_node.lineno, param._ast_node.path) self._params_by_name[param.name] = param def has_documented_type_or_params(self): """Returns whether this type, or any of its parameters, are documented. Use this when deciding whether to create a block of documentation for this type. """ return self.doc or self.has_documented_params() def has_documented_params(self): """Returns whether at least one param is documented.""" return any(param.doc for param in self.params) class Annotation(object): """ Used when a field is annotated with a pre-defined Stone action or a custom annotation. """ def __init__(self, name, namespace, ast_node): self.name = name self.namespace = namespace self._ast_node = ast_node class Deprecated(Annotation): """ Used when a field is annotated for deprecation. """ def __repr__(self): return 'Deprecated(%r, %r)' % (self.name, self.namespace) class Omitted(Annotation): """ Used when a field is annotated for omission. """ def __init__(self, name, namespace, ast_node, omitted_caller): super(Omitted, self).__init__(name, namespace, ast_node) self.omitted_caller = omitted_caller def __repr__(self): return 'Omitted(%r, %r, %r)' % (self.name, self.namespace, self.omitted_caller) class Preview(Annotation): """ Used when a field is annotated for previewing. """ def __repr__(self): return 'Preview(%r, %r)' % (self.name, self.namespace) class Redacted(Annotation): """ Used when a field is annotated for redaction. """ def __init__(self, name, namespace, ast_node, regex=None): super(Redacted, self).__init__(name, namespace, ast_node) self.regex = regex class RedactedBlot(Redacted): """ Used when a field is annotated to be blotted. """ def __repr__(self): return 'RedactedBlot(%r, %r, %r)' % (self.name, self.namespace, self.regex) class RedactedHash(Redacted): """ Used when a field is annotated to be hashed. """ def __repr__(self): return 'RedactedHash(%r, %r, %r)' % (self.name, self.namespace, self.regex) class CustomAnnotation(Annotation): """ Used when a field is annotated with a custom annotation type. """ def __init__(self, name, namespace, ast_node, annotation_type_name, annotation_type_ns, args, kwargs): super(CustomAnnotation, self).__init__(name, namespace, ast_node) self.annotation_type_name = annotation_type_name self.annotation_type_ns = annotation_type_ns self.args = args self.kwargs = kwargs self.annotation_type = None def set_attributes(self, annotation_type): self.annotation_type = annotation_type # check for too many parameters for args if len(self.args) > len(self.annotation_type.params): raise InvalidSpec('Too many parameters passed to annotation type %s' % (self.annotation_type.name), self._ast_node.lineno, self._ast_node.path) # check for unknown keyword arguments acceptable_param_names = set((param.name for param in self.annotation_type.params)) for param_name in self.kwargs: if param_name not in acceptable_param_names: raise InvalidSpec('Unknown parameter %s passed to annotation type %s' % (param_name, self.annotation_type.name), self._ast_node.lineno, self._ast_node.path) for i, param in enumerate(self.annotation_type.params): # first figure out and validate value for this param # arguments are either all kwargs or all args, so don't need to worry about # providing both positional and keyword argument for same parameter if param.name in self.kwargs or i < len(self.args): param_value = self.kwargs[param.name] if self.kwargs else self.args[i] try: param.data_type.check(param_value) except ValueError as e: raise InvalidSpec('Invalid value for parameter %s of annotation type %s: %s' % (param.name, self.annotation_type.name, e), self._ast_node.lineno, self._ast_node.path) elif isinstance(param.data_type, Nullable): param_value = None elif param.has_default: param_value = param.default else: raise InvalidSpec('No value specified for parameter %s of annotation type %s' % (param.name, self.annotation_type.name), self._ast_node.lineno, self._ast_node.path) # now set both kwargs and args to correct value so backend code generators can use # whichever is more convenient (like if kwargs are not supported in a language) self.kwargs[param.name] = param_value if i < len(self.args): self.args[i] = param_value else: self.args.append(param_value) class Alias(Composite): """ NOTE: The categorization of aliases as a composite type is arbitrary. It fit here better than as a primitive or user-defined type. """ def __init__(self, name, namespace, ast_node): """ When this is instantiated, the type is treated as a forward reference. Only when :meth:`set_attributes` is called is the type considered to be fully defined. :param str name: Name of type. :param stone.ir.ApiNamespace namespace: The namespace this type is defined in. :param ast_node: Raw type definition from the parser. :type ast_node: stone.frontend.ast.AstTypeDef """ super(Alias, self).__init__() self._name = name self.namespace = namespace self._ast_node = ast_node # Populated by :meth:`set_attributes` self.raw_doc = None self.doc = None self.data_type = None self.redactor = None self.custom_annotations = [] def set_annotations(self, annotations): for annotation in annotations: if isinstance(annotation, Redacted): # Make sure we don't set multiple conflicting annotations on one alias if self.redactor: raise InvalidSpec("Redactor already set as %r" % str(self.redactor), self._ast_node.lineno) self.redactor = annotation elif isinstance(annotation, CustomAnnotation): # Note: we don't need to do this for builtin annotations because # they are treated as globals at the IR level record_custom_annotation_imports(annotation, self.namespace) self.custom_annotations.append(annotation) else: raise InvalidSpec("Aliases only support 'Redacted' and custom annotations, not %r" % str(annotation), self._ast_node.lineno) def set_attributes(self, doc, data_type): """ :param Optional[str] doc: Documentation string of alias. :param data_type: The source data type referenced by the alias. """ self.raw_doc = doc self.doc = doc_unwrap(doc) self.data_type = data_type # Make sure we don't have a cyclic reference. # Since attributes are set one data type at a time, only the last data # type to be populated in a cycle will be able to detect the cycle. # Before that, the cycle will be broken by an alias with no populated # source. cur_data_type = data_type while is_alias(cur_data_type): cur_data_type = cur_data_type.data_type if cur_data_type == self: raise InvalidSpec( "Alias '%s' is part of a cycle." % self.name, self._ast_node.lineno, self._ast_node.path) @property def name(self): return self._name def check(self, val): return self.data_type.check(val) def check_example(self, ex_field): # TODO: Assert that this isn't a user-defined type. return self.data_type.check_example(ex_field) def _has_example(self, label): # TODO: Assert that this is a user-defined type return self.data_type._has_example(label) def _compute_example(self, label): return self.data_type._compute_example(label) def check_attr_repr(self, attr_field): return self.data_type.check_attr_repr(attr_field) def __repr__(self): return 'Alias(%r, %r)' % (self.name, self.data_type) def unwrap_nullable(data_type): """ Convenience method to unwrap Nullable from around a DataType. Args: data_type (DataType): The target to unwrap. Return: Tuple[DataType, bool]: The underlying data type and a bool indicating whether the input type was nullable. """ if is_nullable_type(data_type): return data_type.data_type, True else: return data_type, False def unwrap_aliases(data_type): """ Convenience method to unwrap all Alias(es) from around a DataType. Args: data_type (DataType): The target to unwrap. Return: Tuple[DataType, bool]: The underlying data type and a bool indicating whether the input type had at least one alias layer. """ unwrapped_alias = False while is_alias(data_type): unwrapped_alias = True data_type = data_type.data_type return data_type, unwrapped_alias def resolve_aliases(data_type): """ Resolve all chained / nested aliases. This will recursively point nested aliases to their resolved data type (first non-alias in the chain). Note: This differs from unwrap_alias which simply identifies/returns the resolved data type. Args: data_type (DataType): The target DataType/Alias to resolve. Return: DataType: The resolved type. """ if not is_alias(data_type): return data_type resolved = resolve_aliases(data_type.data_type) data_type.data_type = resolved return resolved def strip_alias(data_type): """ Strip alias from a data_type chain - this function should be used *after* aliases are resolved (see resolve_aliases fn): Loops through given data type chain (unwraps types), replaces first alias with underlying type, and then terminates. Note: Stops on encountering the first alias as it assumes intermediate aliases are already removed. Args: data_type (DataType): The target DataType chain to strip. Return: None """ while hasattr(data_type, 'data_type'): if is_alias(data_type.data_type): data_type.data_type = data_type.data_type.data_type break data_type = data_type.data_type def unwrap(data_type): """ Convenience method to unwrap all Aliases and Nullables from around a DataType. This checks for nullable wrapping aliases, as well as aliases wrapping nullables. Args: data_type (DataType): The target to unwrap. Return: Tuple[DataType, bool, bool]: The underlying data type; a bool that is set if a nullable was present; a bool that is set if an alias was present. """ unwrapped_nullable = False unwrapped_alias = False while is_alias(data_type) or is_nullable_type(data_type): if is_nullable_type(data_type): unwrapped_nullable = True if is_alias(data_type): unwrapped_alias = True data_type = data_type.data_type return data_type, unwrapped_nullable, unwrapped_alias def is_alias(data_type): return isinstance(data_type, Alias) def is_bytes_type(data_type): return isinstance(data_type, Bytes) def is_boolean_type(data_type): return isinstance(data_type, Boolean) def is_composite_type(data_type): return isinstance(data_type, Composite) def is_field_type(data_type): return isinstance(data_type, Field) def is_float_type(data_type): return isinstance(data_type, (Float32, Float64)) def is_integer_type(data_type): return isinstance(data_type, (UInt32, UInt64, Int32, Int64)) def is_list_type(data_type): return isinstance(data_type, List) def is_map_type(data_type): return isinstance(data_type, Map) def is_nullable_type(data_type): return isinstance(data_type, Nullable) def is_numeric_type(data_type): return is_integer_type(data_type) or is_float_type(data_type) def is_primitive_type(data_type): return isinstance(data_type, Primitive) def is_string_type(data_type): return isinstance(data_type, String) def is_struct_type(data_type): return isinstance(data_type, Struct) def is_tag_ref(val): return isinstance(val, TagRef) def is_timestamp_type(data_type): return isinstance(data_type, Timestamp) def is_union_type(data_type): return isinstance(data_type, Union) def is_user_defined_type(data_type): return isinstance(data_type, UserDefined) def is_void_type(data_type): return isinstance(data_type, Void) def is_int32_type(data_type): return isinstance(data_type, Int32) def is_int64_type(data_type): return isinstance(data_type, Int64) def is_uint32_type(data_type): return isinstance(data_type, UInt32) def is_uint64_type(data_type): return isinstance(data_type, UInt64) def is_float32_type(data_type): return isinstance(data_type, Float32) def is_float64_type(data_type): return isinstance(data_type, Float64)