in src/python/pants/util/objects.py [0:0]
def datatype(field_decls, superclass_name=None, **kwargs):
"""A wrapper for `namedtuple` that accounts for the type of the object in equality.
Field declarations can be a string, which declares a field with that name and
no type checking. Field declarations can also be a tuple `('field_name',
field_type)`, which declares a field named `field_name` which is type-checked
at construction. If a type is given, the value provided to the constructor for
that field must be exactly that type (i.e. `type(x) == field_type`), and not
e.g. a subclass.
:param field_decls: Iterable of field declarations.
:return: A type object which can then be subclassed.
:raises: :class:`TypeError`
"""
field_names = []
fields_with_constraints = OrderedDict()
for maybe_decl in field_decls:
# ('field_name', type)
if isinstance(maybe_decl, tuple):
field_name, type_spec = maybe_decl
if isinstance(type_spec, type):
type_constraint = Exactly(type_spec)
elif isinstance(type_spec, TypeConstraint):
type_constraint = type_spec
else:
raise TypeError(
"type spec for field '{}' was not a type or TypeConstraint: was {!r} (type {!r})."
.format(field_name, type_spec, type(type_spec).__name__))
fields_with_constraints[field_name] = type_constraint
else:
# interpret it as a field name without a type to check
field_name = maybe_decl
# namedtuple() already checks field uniqueness
field_names.append(field_name)
if not superclass_name:
superclass_name = '_anonymous_namedtuple_subclass'
namedtuple_cls = namedtuple(superclass_name, field_names, **kwargs)
class DataType(namedtuple_cls, DatatypeMixin):
type_check_error_type = TypedDatatypeInstanceConstructionError
def __new__(cls, *args, **kwargs):
# TODO: Ideally we could execute this exactly once per `cls` but it should be a
# relatively cheap check.
if not hasattr(cls.__eq__, '_eq_override_canary'):
raise cls.make_type_error('Should not override __eq__.')
try:
this_object = super(DataType, cls).__new__(cls, *args, **kwargs)
except TypeError as e:
raise cls.make_type_error(
"error in namedtuple() base constructor: {}".format(e))
# TODO: Make this kind of exception pattern (filter for errors then display them all at once)
# more ergonomic.
type_failure_msgs = []
for field_name, field_constraint in fields_with_constraints.items():
field_value = getattr(this_object, field_name)
try:
field_constraint.validate_satisfied_by(field_value)
except TypeConstraintError as e:
type_failure_msgs.append(
"field '{}' was invalid: {}".format(field_name, e))
if type_failure_msgs:
raise cls.make_type_error(
'{} type checking constructor arguments:\n{}'
.format(pluralize(len(type_failure_msgs), 'error'),
'\n'.join(type_failure_msgs)))
return this_object
def __eq__(self, other):
if self is other:
return True
# Compare types and fields.
if type(self) != type(other):
return False
# Explicitly return super.__eq__'s value in case super returns NotImplemented
return super(DataType, self).__eq__(other)
# We define an attribute on the `cls` level definition of `__eq__` that will allow us to detect
# that it has been overridden.
__eq__._eq_override_canary = None
def __ne__(self, other):
return not (self == other)
# NB: in Python 3, whenever __eq__ is overridden, __hash__() must also be
# explicitly implemented, otherwise Python will raise "unhashable type". See
# https://docs.python.org/3/reference/datamodel.html#object.__hash__.
def __hash__(self):
try:
return super(DataType, self).__hash__()
except TypeError:
# If any fields are unhashable, we want to be able to specify which ones in the error
# message, but we don't want to slow down the normal __hash__ code path, so we take the time
# to break it down by field if we know the __hash__ fails for some reason.
for field_name, value in self._asdict().items():
try:
hash(value)
except TypeError as e:
raise TypeError("For datatype object {} (type '{}'): in field '{}': {}"
.format(self, type(self).__name__, field_name, e))
# If the error doesn't seem to be with hashing any of the fields, just re-raise the
# original error.
raise
# NB: As datatype is not iterable, we need to override both __iter__ and all of the
# namedtuple methods that expect self to be iterable.
def __iter__(self):
raise self.make_type_error("datatype object is not iterable")
def _super_iter(self):
return super(DataType, self).__iter__()
def _asdict(self):
"""Return a new OrderedDict which maps field names to their values.
Overrides a namedtuple() method which calls __iter__.
"""
return OrderedDict(zip(self._fields, self._super_iter()))
def _replace(self, **kwargs):
"""Return a new datatype object replacing specified fields with new values.
Overrides a namedtuple() method which calls __iter__.
"""
field_dict = self._asdict()
field_dict.update(**kwargs)
return type(self)(**field_dict)
def copy(self, **kwargs):
return self._replace(**kwargs)
# NB: it is *not* recommended to rely on the ordering of the tuple returned by this method.
def __getnewargs__(self):
"""Return self as a plain tuple. Used by copy and pickle."""
return tuple(self._super_iter())
def __repr__(self):
args_formatted = []
for field_name in field_names:
field_value = getattr(self, field_name)
args_formatted.append("{}={!r}".format(field_name, field_value))
return '{class_name}({args_joined})'.format(
class_name=type(self).__name__,
args_joined=', '.join(args_formatted))
def __str__(self):
elements_formatted = []
for field_name in field_names:
constraint_for_field = fields_with_constraints.get(field_name, None)
field_value = getattr(self, field_name)
if not constraint_for_field:
elements_formatted.append(
# TODO: consider using the repr of arguments in this method.
"{field_name}={field_value}"
.format(field_name=field_name,
field_value=field_value))
else:
elements_formatted.append(
"{field_name}<{type_constraint}>={field_value}"
.format(field_name=field_name,
type_constraint=constraint_for_field,
field_value=field_value))
return '{class_name}({typed_tagged_elements})'.format(
class_name=type(self).__name__,
typed_tagged_elements=', '.join(elements_formatted))
# Return a new type with the given name, inheriting from the DataType class
# just defined, with an empty class body.
try: # Python3
return type(superclass_name, (DataType,), {})
except TypeError: # Python2
return type(superclass_name.encode('utf-8'), (DataType,), {})