Repository URL to install this package:
|
Version:
1.26.0.dev0+gite506aa5f ▾
|
# Copyright 2015 Pants project contributors (see CONTRIBUTORS.md).
# Licensed under the Apache License, Version 2.0 (see LICENSE).
from collections.abc import MutableMapping, MutableSequence
from typing import Any, Dict, Iterable, Optional, cast
from pants.build_graph.address import Address
from pants.engine.addressable import addressable, addressable_list
from pants.engine.objects import Serializable, SerializableFactory, Validatable, ValidationError
from pants.util.objects import SubclassesOf, SuperclassesOf
class Struct(Serializable, SerializableFactory, Validatable):
"""A serializable object.
A Struct is composed of basic python builtin types and other high-level Structs. Structs can
carry a name in which case they become addressable and can be reused.
"""
# Fields dealing with inheritance.
_INHERITANCE_FIELDS = {"extends", "merges"}
# The type alias for an instance overwrites any inherited type_alias field.
_TYPE_ALIAS_FIELD = "type_alias"
# The field that indicates whether a Struct is abstract (and should thus skip validation).
_ABSTRACT_FIELD = "abstract"
# Fields that should not be inherited.
_UNINHERITABLE_FIELDS = _INHERITANCE_FIELDS | {_TYPE_ALIAS_FIELD, _ABSTRACT_FIELD}
# Fields that are only intended for consumption by the Struct baseclass.
_INTERNAL_FIELDS = _INHERITANCE_FIELDS | {_ABSTRACT_FIELD}
def __init__(
self,
abstract: bool = False,
extends: Optional[Serializable] = None,
merges: Optional[Iterable[Serializable]] = None,
type_alias: Optional[str] = None,
**kwargs: Any,
) -> None:
"""Creates a new struct data blob.
By default Structs are anonymous (un-named), concrete (not `abstract`), and they neither
inherit nor merge another Struct.
Inheritance is allowed via the `extends` and `merges` channels. An object inherits all
attributes from the object it extends, overwriting any attributes in common with the extended
object with its own. The relationship is an "overlay". For the merges, the same rules apply
for as for extends working left to right such that the rightmost merges attribute will overwrite
any similar attribute from merges to its left where the main object does not itself define the
attribute. The primary difference is in handling of lists and dicts. These are merged and not
over-written; again working from left to right with the main object's collection serving as the
seed when present.
A Struct can be semantically abstract without setting `abstract=True`. The `abstract`
value can serve as documentation, or, for subclasses that provide an implementation for
`validate_concrete`, it allows skipping validation for abstract instances.
:param abstract: `True` to mark this struct as abstract, in which case no
validation is performed (see `validate_concrete`); `False` by default.
:param extends: The struct instance to inherit field values from. Any shared fields are
over-written with this instances values.
:type extends: An addressed or concrete struct instance that is a type compatible with
this struct or this structs superclasses.
:param merges: The struct instances to merge this instance's field values with. Merging
is like extension except for containers, which are extended instead of replaced;
ie: any `dict` values are updated with this instances items and any `list` values
are extended with this instances items.
:type merges: An addressed or concrete struct instance that is a type compatible with
this struct or this structs superclasses.
:param **kwargs: The struct parameters.
"""
self._kwargs = kwargs
self._kwargs["abstract"] = abstract
self._kwargs[self._TYPE_ALIAS_FIELD] = type_alias
self.extends = extends
self.merges = merges
# Allow for structs that are directly constructed in memory. These can have an
# address directly assigned (vs. inferred from name + source file location) and we only require
# that if they do, their name - if also assigned, matches the address.
if self.address:
target_name, _, config_specifier = self.address.target_name.partition("@")
if self.name and self.name != target_name:
self.report_validation_error(
"Address and name do not match! address: {}, name: {}".format(
self.address, self.name
)
)
self._kwargs["name"] = target_name
def kwargs(self) -> Dict[str, Any]:
"""Returns a dict of the kwargs for this Struct which were not interpreted by the baseclass.
This excludes fields like `extends`, `merges`, and `abstract`, which are consumed by
SerializableFactory.create and Validatable.validate.
"""
return {k: v for k, v in self._kwargs.items() if k not in self._INTERNAL_FIELDS}
@property
def name(self) -> Optional[str]:
"""Return the name of this object, if any.
In general structs need not be named, in which case they are generally embedded
objects; ie: attributes values of enclosing named structs. Any top-level
struct object, though, will carry a unique name (in the struct object's enclosing
namespace) that can be used to address it.
:rtype: string
"""
return self._kwargs.get("name")
@property
def address(self) -> Optional[Address]:
"""Return the address of this object, if any.
In general structs need not be identified by an address, in which case they are generally
embedded objects; ie: attributes values of enclosing named structs. Any top-level struct,
though, will be identifiable via a unique address.
"""
return cast(Optional[Address], self._kwargs.get("address"))
@property
def type_alias(self) -> str:
"""Return the type alias this target was constructed via.
For a target read from a BUILD file, this will be target alias, like 'java_library'.
For a target constructed in memory, this will be the simple class name, like 'JavaLibrary'.
The end result is that the type alias should be the most natural way to refer to this target's
type to the author of the target instance.
"""
type_alias: Optional[str] = self._kwargs.get(self._TYPE_ALIAS_FIELD, None)
return type_alias if type_alias is not None else type(self).__name__
@property
def abstract(self) -> bool:
"""Return `True` if this object has been marked as abstract.
Abstract objects are not validated. See: `validate_concrete`.
"""
return self._kwargs.get("abstract", False)
# It only makes sense to inherit a subset of our own fields (we should not inherit new fields!),
# our superclasses logically provide fields within this constrained set.
# NB: Since `Struct` is at base an ~unconstrained struct, a superclass does allow for
# arbitrary and thus more fields to be defined than a subclass might logically support. We
# accept this hole in a trade for generally expected behavior when `Struct` is subclassed
# in the style of constructors with named parameters representing the full complete set of
# expected parameters leaving **kwargs only for use by 'the system'; ie for `type_alias` and
# `address` plumbing for example.
#
# Of note is the fact that we pass a constraint type and not a concrete constraint value. This
# tells addressable to use `SuperclassesOf([Struct instance's type])`, which is what we
# want. Aka, for `StructSubclassA`, the constraint is
# `SuperclassesOf(StructSubclassA)`.
#
@addressable(SuperclassesOf)
def extends(self):
"""Return the object this object extends, if any.
:rtype: :class:`Serializable`
"""
@addressable_list(SuperclassesOf)
def merges(self):
"""Return the objects this object merges in, if any.
:rtype: list of :class:`Serializable`
"""
def _asdict(self) -> Dict[str, Any]:
return self._kwargs
def _extract_inheritable_attributes(self, serializable):
attributes = serializable._asdict().copy()
# Allow for embedded objects inheriting from addressable objects - they should never inherit an
# address and any top-level object inheriting will have its own address.
attributes.pop("address", None)
# We should never inherit special fields - these are for local book-keeping only.
for field in self._UNINHERITABLE_FIELDS:
attributes.pop(field, None)
return attributes
def create(self) -> "Struct":
if not (self.extends or self.merges):
return self
# Filter out the attributes that we will consume below for inheritance.
attributes = {
k: v
for k, v in self._asdict().items()
if k not in self._INHERITANCE_FIELDS and v is not None
}
if self.extends:
for k, v in self._extract_inheritable_attributes(self.extends).items():
attributes.setdefault(k, v)
if self.merges:
def merge(attrs):
for k, v in attrs.items():
if isinstance(v, MutableMapping):
mapping = attributes.get(k, {})
mapping.update(v)
attributes[k] = mapping
elif isinstance(v, MutableSequence):
sequence = attributes.get(k, [])
sequence.extend(v)
attributes[k] = sequence
else:
attributes.setdefault(k, v)
for merged in self.merges:
merge(self._extract_inheritable_attributes(merged))
struct_type = type(self)
return struct_type(**attributes)
def validate(self) -> None:
if not self.abstract:
self.validate_concrete()
def report_validation_error(self, message):
"""Raises a properly identified validation error.
:param string message: An error message describing the validation error.
:raises: :class:`pants.engine.objects.ValidationError`
"""
raise ValidationError(self.address, message)
def validate_concrete(self) -> None:
"""Subclasses can override to implement validation logic.
The object will be fully hydrated state and it's guaranteed the object will be concrete, aka.
not `abstract`. If an error is found in the struct's fields, a validation error should
be raised by calling `report_validation_error`.
:raises: :class:`pants.engine.objects.ValidationError`
"""
def __getattr__(self, item):
if item in self._kwargs:
return self._kwargs[item]
# NB: This call ensures that the default missing attribute behavior happens.
# Without it, AttributeErrors inside @property methods will be misattributed.
return object.__getattribute__(self, item)
def _key(self):
def hashable(value):
if isinstance(value, dict):
return tuple(sorted((k, hashable(v)) for k, v in value.items()))
elif isinstance(value, list):
return tuple(hashable(v) for v in value)
elif isinstance(value, set):
return tuple(sorted(hashable(v) for v in value))
else:
return value
return tuple(
sorted(
(k, hashable(v))
for k, v in self._kwargs.items()
if k not in self._INHERITANCE_FIELDS
)
)
def __hash__(self):
return hash(self._key())
def __eq__(self, other):
return isinstance(other, Struct) and self._key() == other._key()
def __ne__(self, other):
return not (self == other)
def __repr__(self):
classname = type(self).__name__
if self.address:
return "{classname}(address={address})".format(
classname=classname, address=self.address.reference()
)
else:
return "{classname}({args})".format(
classname=classname,
args=", ".join(
sorted("{}={!r}".format(k, v) for k, v in self._kwargs.items() if v)
),
)
class StructWithDeps(Struct):
"""A subclass of Struct with dependencies."""
def __init__(self, dependencies=None, **kwargs):
"""
:param list dependencies: The direct dependencies of this struct.
"""
# TODO: enforce the type of variants using the Addressable framework.
super().__init__(**kwargs)
self.dependencies = dependencies
@addressable_list(SubclassesOf(Struct))
def dependencies(self):
"""The direct dependencies of this target.
:rtype: list
"""