Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
except ProxyTypeError:
raise ProxyTypeError(
"Expected Dict values of type {}, but got {}".format(vt, val)
)
if is_str_dict:
promoted[key] = promoted_val
# note we use the unpromoted key, which should be a string
# this is an optimization that produces a cleaner graph for the case of string-keyed dicts
else:
promoted += [promoted_key, promoted_val]
# for non-string dicts, we just give varargs of key, value, key, value, ...
# since that's a much simpler graft representation than constructing a list
# of tuples
if is_str_dict:
self.graft = client.apply_graft("dict.create", **promoted)
else:
self.graft = client.apply_graft("dict.create", *promoted)
def __init__(self, iterable):
if self._type_params is None:
raise TypeError(
"Cannot instantiate a generic List; the item type must be specified (like `List[Int]`)"
)
if isinstance(iterable, type(self)):
self.graft = client.apply_graft("list.copy", iterable)
elif isinstance(iterable, List):
raise ProxyTypeError(
"Cannot convert {} to {}, since they have different value types".format(
type(iterable).__name__, type(self).__name__
)
)
else:
if not isinstance(iterable, abc.Iterable):
raise ProxyTypeError("Expected an iterable, got {}".format(iterable))
value_type = self._type_params[0]
def checker_promoter(i, x):
try:
return value_type._promote(x)
except ProxyTypeError:
raise ProxyTypeError(
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._items_cache = promoted
# ^ this _might_ be wrong, since the getattr graft won't include `getattr`
)
)
def checker_promoter(i, x):
cls = value_types[i]
try:
return cls._promote(x)
except ProxyTypeError:
raise ProxyTypeError(
"While constructing {}, expected {} for tuple element {}, but got {!r}".format(
type(self).__name__, cls, i, x
)
)
iterable = tuple(checker_promoter(i, x) for i, x in enumerate(iterable))
self.graft = client.apply_graft("tuple", *iterable)
result = func(**args)
if returns is not None:
try:
result = returns._promote(result)
except ProxyTypeError as e:
raise ProxyTypeError(
"Cannot promote {} to {}, the expected return type of the function: {}".format(
result, returns.__name__, e
)
)
else:
result = proxify(result)
return type(result)._from_graft(
client.function_graft(
result, *tuple(func_signature.parameters), first_guid=first_guid
)
def object(self):
"Proxytype: The proxy object this Job computes."
if self._object is None:
typespec = json.loads(self._message.serialized_typespec)
proxytype = deserialize_typespec(typespec)
graft = json.loads(self._message.serialized_graft)
isolated = graft_client.isolate_keys(graft)
self._object = proxytype._from_graft(isolated)
return self._object
def __init__(self, images):
"Construct an ImageCollection from a sequence of Images"
self.graft = client.apply_graft(
"ImageCollection.from_images", images, env.geoctx
)
def _from_proto(cls, message, client=None):
typespec = json.loads(message.serialized_typespec)
proxytype = deserialize_typespec(typespec)
if message.serialized_graft:
graft = json.loads(message.serialized_graft)
isolated = graft_client.isolate_keys(graft)
obj = proxytype._from_graft(isolated)
else:
raise AttributeError(
(
"The serialized graft attribute does not exist or "
"acces is not authorized for XYZ '{}'. To share "
"objects with others, please use a Workflow instead."
).format(message.id)
)
return cls(obj, message, client=client)
def __init__(self, value):
self.graft = client.value_graft(value)
def _from_proto(cls, message, client=None):
typespec = json.loads(message.serialized_typespec)
proxytype = deserialize_typespec(typespec)
if message.serialized_graft:
graft = json.loads(message.serialized_graft)
isolated = graft_client.isolate_keys(graft)
obj = proxytype._from_graft(isolated)
else:
obj = proxytype._from_apply("Workflow.use", workflow_id=message.id)
return cls(obj, message, client=client)