Source code for descarteslabs.workflows.types.containers.struct

from ....common.graft import client
from ...cereal import serializable
from ..core import ProxyTypeError, GenericProxytype, assert_is_proxytype, merge_params
from ..primitives import NoneType, Any

def _dict_repr_to_constructor_syntax(string):
    In string, convert literal dict reprs like ``{'x': 1}`` to constructors like ``dict(x=1)``.

    Only works for dicts without string values (which is fine for a typespec, where values are always Proxytypes)
    return (
        string.replace("{", "dict(")
        .replace("}", ")")
        .replace(": ", "=")
        .replace("'", "")

def _property_factory(field, field_cls, doc=None):
    def _getter(self):
        return self._attr(field)

    field_cls_name = field_cls.__name__
    if "{" in field_cls_name:
        # NOTE(gabe): Work around a bizarre `sphinx.ext.napoleon` bug,
        # which ignores inline markup (like ``) when searching for colons,
        # causing it to split the type and message parts at the first colon
        # in a dict's repr. Converting to `dict(x=1)`-like syntax seemed to be
        # the best way to keep unmangled docstrings for IPython while remaining
        # sphinx-compatible
        field_cls_name = _dict_repr_to_constructor_syntax(field_cls_name)
    if doc is not None:
        doc = "{}: {}".format(field_cls_name, doc)
        doc = field_cls_name

    _getter.__doc__ = doc
    _getter.__name__ = field
    return property(_getter)

[docs]@serializable() class Struct(GenericProxytype): """ ``Struct[{field_name: type, ...}]``: Proxy container with named fields of specific types, meant as a helper base class. Can be instantiated from kwargs. In general, Struct is used as an internal base class to help create proxy objects, and is not meant to be worked with directly. Notes ----- Adding certain fields to subclasses of concrete Structs changes their behavior: * ``_doc`` (dict of str): Docstrings for struct fields. The keys must match keys in the typespec (though not all fields must be documented). Resulting docstrings are formatted like ``"{}: {}".format(return_type_name, given_docstring)``. If no docstring is given, the field getter will just be documented with its return type. * ``_constructor`` (str): Function name to use in graft for instantiation * ``_optional`` (set of str, or None): Optional field names for constructor * ``_read_only``: (set of str, or None): Field names that can't be given to constructor Examples -------- >>> from descarteslabs.workflows import Struct, Str, Int >>> MyStructType = Struct[{'x': Str, 'y': Int}] # struct where type of 'x' is Str and type of 'y' is Int >>> instance = MyStructType(x="foo", y=1) >>> instance <descarteslabs.workflows.types.containers.struct.Struct[{'x': Str, 'y': Int}] object at 0x...> >>> from descarteslabs.workflows import Struct, Str, Int >>> class Foo(Struct[{'x': Int, 'y': Str}]): # doctest: +SKIP ... _doc = { ... 'x': "the x field", ... 'y': "the y field, derived from x", ... } ... _constructor = "make_foo" ... _optional = {'x'} ... _read_only = {'y'} >>> my_struct = Foo(10) # doctest: +SKIP >>> my_struct.x.compute() # doctest: +SKIP 10 """ # docstrings for struct fields; must be dict[str, str] or None _doc = None # function name to use in graft for instantiation; can be overridden in subclasses _constructor = "wf.struct" # optional arguments for __init__; must be a set or None _optional = None # read only attributes _read_only = None @staticmethod def __class_getitem_hook__(name, bases, dct, type_params): # Called when constructing a concrete subtype (`Struct[{'x': Int}]`). # We add `@property` getter functions for each field in the type params, # which makes the struct fields easier to document and introspect. dct.update( { field: _property_factory(field, field_cls) for field, field_cls in type_params[0].items() } ) return name, bases, dct @classmethod def __init_subclass__(subcls, **kwargs): super().__init_subclass__(**kwargs) # If a subclass of a Struct sets the `_doc` class attribute, # we replace any field getters (created in `__class_getitem_hook__`) # with new field getters that include the given docstrings try: type_params = subcls._type_params[0] except IndexError: return docs = subcls._doc if docs is not None: for field, docstring in docs.items(): if field not in type_params: raise TypeError( "Cannot document field {!r}: it does not exist in {}".format( field, subcls ) ) field_cls = type_params[field] setattr(subcls, field, _property_factory(field, field_cls, docstring)) def __new__(cls, *args, **kwargs): new = super(Struct, cls).__new__(cls) # Always create an `_items_cache` cache of looked-up attrs, # even when __init__ is bypassed by `from_graft` new._items_cache = {} return new # QUESTION(gabe): I'm not sure having __init__ take (and require) kwargs # for all the fields is the best move, since subclasses (which represent user-facing objects) # may want very different constructors, or to be cast-only and not even have a public constructor. # We'll see. def __init__(self, **kwargs): if self._type_params is None: raise TypeError( "Cannot instantiate a generic {}; the item types must be specified " "(like Struct[{}])".format(type(self).__name__, "{'a': Str, 'b': Int}") ) promoted = self._promote_kwargs( kwargs, optional=self._optional, read_only=self._read_only ) self.graft = client.apply_graft(self._constructor, **promoted) self.params = merge_params(self._constructor, *kwargs.values()) # ^ NOTE: this would need to include the keys as well if proxytypes ever become hashable self._items_cache = promoted # ^ this _might_ be wrong, since the getattr graft won't include `getattr` # and just directly reference the promoted object, but that seems ok (while everything is immutable) @classmethod def _validate_params(cls, type_params): for type_param in type_params: assert isinstance( type_param, dict ), "Struct type parameters must be specified with a dictionary of field name to type" for key, param_cls in type_param.items(): assert isinstance( key, str ), "Field names must be strings, but '{}' is a {!r}".format( key, type(key) ) error_message = "Struct item type parameters must be Proxytypes but for field '{}', got {}".format( key, param_cls ) assert_is_proxytype(param_cls, error_message=error_message) @classmethod def _promote(cls, obj): if isinstance(obj, cls): return obj if isinstance(obj, Any): return obj.cast(cls) else: raise ProxyTypeError("Cannot promote {} to {}".format(obj, cls)) from None @classmethod def _promote_kwargs(cls, kwargs, optional=None, read_only=None): if optional is None: optional = set() if not isinstance(optional, set): raise TypeError( "Optional kwargs must be given as a set, not {}".format(optional) ) if read_only is None: read_only = set() if not isinstance(read_only, set): raise TypeError( "Read only kwargs must be given as a set, not {}".format(read_only) ) class_name = cls.__name__ try: type_params = cls._type_params[0] except TypeError: raise TypeError( "Cannot instantiate a generic {}; the item types must be specified".format( class_name ) ) missing_args = type_params.keys() - kwargs.keys() - optional - read_only if len(missing_args) > 0: raise ProxyTypeError( "Missing required keyword arguments to {}: {}".format( class_name, ", ".join(map(repr, missing_args)) ) ) provided_read_only_args = kwargs.keys() & read_only if len(provided_read_only_args) > 0: raise ProxyTypeError( "Read only keyword argument to {}: {}".format( class_name, ", ".join(map(repr, provided_read_only_args)) ) ) promoted_kwargs = {} for field_name, val in kwargs.items(): try: field_cls = type_params[field_name] except KeyError: raise ProxyTypeError( "{} has no field {!r}".format(class_name, field_name) ) if val is None or isinstance(val, NoneType) and field_name in optional: continue try: promoted_val = field_cls._promote(val) except ProxyTypeError as e: raise ProxyTypeError( "In field {!r} of {}, expected {}, but got {}: {}".format( field_name, class_name, field_cls, type(val), val ), e, ) promoted_kwargs[field_name] = promoted_val return promoted_kwargs def _attr(self, attr): try: item = self._items_cache[attr] except KeyError: try: value_type = self._type_params[0][attr] except KeyError: raise AttributeError( "{} has no attribute {!r}".format(type(self).__name__, attr) ) item = value_type._from_apply("wf.getattr", self, attr) self._items_cache[attr] = item return item