Source code for

# stdlib
from typing import Any
from typing import List
from typing import Optional

# third party
from google.protobuf.reflection import GeneratedProtocolMessageType

# syft absolute
import syft as sy

# syft relative
from ...logger import traceback_and_raise
from import StorableObject as StorableObject_PB
from ...util import get_fully_qualified_name
from ...util import index_syft_by_module_name
from ...util import key_emoji
from ..common.serde.deserialize import _deserialize
from ..common.serde.serializable import Serializable
from ..common.serde.serializable import bind_protobuf
from ..common.storeable_object import AbstractStorableObject
from ..common.uid import UID

[docs]@bind_protobuf class StorableObject(AbstractStorableObject): """ StorableObject is a wrapper over some Serializable objects, which we want to keep in an ObjectStore. The Serializable objects that we want to store have to be backed up in syft-proto in the StorableObject protobuffer, where you can find more details on how to add new types to be serialized. This object is frozen, you cannot change one in place. Arguments: id (UID): the id at which to store the data. data (Serializable): A serializable object. description (Optional[str]): An optional string that describes what you are storing. Useful when searching. tags (Optional[List[str]]): An optional list of strings that are tags used at search. TODO: add docs about read_permission and search_permission Attributes: id (UID): the id at which to store the data. data (Serializable): A serializable object. description (Optional[str]): An optional string that describes what you are storing. Useful when searching. tags (Optional[List[str]]): An optional list of strings that are tags used at search. """ __slots__ = ["id", "_data", "_description", "_tags"] def __init__( self, id: UID, data: object, description: Optional[str] = None, tags: Optional[List[str]] = None, read_permissions: Optional[dict] = None, search_permissions: Optional[dict] = None, ): = id = data self._description: str = description if description else "" self._tags: List[str] = tags if tags else [] # the dict key of "verify key" objects corresponding to people # the value is the original request_id to allow lookup later # who are allowed to call .get() and download this object. self.read_permissions = read_permissions if read_permissions else {} # the dict key of "verify key" objects corresponding to people # the value is the original request_id to allow lookup later # who are allowed to know that the tensor exists (via search or other means) self.search_permissions: dict = search_permissions if search_permissions else {} @property def object_type(self) -> str: return str(type( @property def object_qualname(self) -> str: return get_fully_qualified_name( # Why define data as a property? # For C type/class objects as data. # We need to use it's wrapper type very often inside StorableObject, so we set _data # attribute as it's wrapper object. But we still want to give a straight API to users, # so we return the initial C type object when user call # For python class objects as data. data and _data are the same thing. @property # type: ignore def data(self) -> Any: # type: ignore if type(self._data).__name__.endswith("Wrapper"): return self._data.obj else: return self._data @data.setter def data(self, value: Any) -> Any: if hasattr(value, "_sy_serializable_wrapper_type"): self._data = value._sy_serializable_wrapper_type(value=value) else: self._data = value @property def tags(self) -> Optional[List[str]]: return self._tags @tags.setter def tags(self, value: Optional[List[str]]) -> None: self._tags = value if value else [] @property def description(self) -> Optional[str]: return self._description @description.setter def description(self, description: Optional[str]) -> None: self._description = description if description else "" def _object2proto(self) -> StorableObject_PB: proto = StorableObject_PB() # Step 1: Serialize the id to protobuf and copy into protobuf id = sy.serialize( # Step 2: Save the type of wrapper to use to deserialize proto.data_type = get_fully_qualified_name(obj=self._data) # Step 3: Serialize data to protobuf and pack into proto data = self._data._object2proto() if hasattr(self, "description"): # Step 4: save the description into proto proto.description = self.description # QUESTION: Which one do we want, or self.tags or both??? if hasattr(self, "tags"): # Step 5: save tags into proto if they exist if self.tags is not None: for tag in self.tags: proto.tags.append(tag) # Step 6: save read permissions if len(self.read_permissions) > 0: permission_data = sy.lib.python.Dict() for k, v in self.read_permissions.items(): permission_data[k] = v proto.read_permissions = sy.serialize(permission_data, to_bytes=True) # Step 7: save search permissions if len(self.search_permissions.keys()) > 0: permission_data = sy.lib.python.Dict() for k, v in self.search_permissions.items(): permission_data[k] = v proto.search_permissions = sy.serialize(permission_data, to_bytes=True) return proto @staticmethod def _proto2object(proto: StorableObject_PB) -> Serializable: # Step 1: deserialize the ID id = _deserialize( if not isinstance(id, UID): traceback_and_raise(ValueError("TODO")) # Step 2: get the type of wrapper to use to deserialize data_type = index_syft_by_module_name(fully_qualified_name=proto.data_type) # Step 3: get the protobuf type we deserialize for .data schematic_type = data_type.get_protobuf_schema() # type: ignore # Step 4: Deserialize data from protobuf data = None if callable(schematic_type): data = schematic_type() descriptor = getattr(schematic_type, "DESCRIPTOR", None) if descriptor is not None and data = data_type._proto2object(proto=data) # type: ignore # Step 5: get the description from proto description = proto.description if proto.description else "" # Step 6: get the tags from proto of they exist tags = list(proto.tags) if proto.tags else [] # Step 7: get the read permissions read_permissions = None if proto.read_permissions is not None and len(proto.read_permissions) > 0: read_permissions = _deserialize( blob=proto.read_permissions, from_bytes=True ) # Step 8: get the search permissions search_permissions = None if proto.search_permissions is not None and len(proto.search_permissions) > 0: search_permissions = _deserialize( blob=proto.search_permissions, from_bytes=True ) result = StorableObject( id=id, data=data, description=description, tags=tags, read_permissions=read_permissions, search_permissions=search_permissions, ) return result
[docs] @staticmethod def get_protobuf_schema() -> GeneratedProtocolMessageType: """Return the type of protobuf object which stores a class of this type As a part of serialization and deserialization, we need the ability to lookup the protobuf object type directly from the object type. This static method allows us to do this. Importantly, this method is also used to create the reverse lookup ability within the metaclass of Serializable. In the metaclass, it calls this method and then it takes whatever type is returned from this method and adds an attribute to it with the type of this class attached to it. See the MetaSerializable class for details. :return: the type of protobuf object which corresponds to this class. :rtype: GeneratedProtocolMessageType """ return StorableObject_PB
def __repr__(self) -> str: return ( "<Storable: " +"\n", "").replace(" ", " ") + ">" ) @property def icon(self) -> str: return "🗂️" @property def pprint(self) -> str: output = f"{self.icon} ({self.class_name}) (" if hasattr(, "pprint"): output += elif is not None: output += else: output += "(Key Only)" if len(self._description) > 0: output += f" desc: {self.description}" if len(self._tags) > 0: output += f" tags: {self.tags}" if len(self.read_permissions.keys()) > 0: output += ( " can_read: " + f"{[key_emoji(key=key) for key in self.read_permissions.keys()]}" ) if len(self.search_permissions.keys()) > 0: output += ( " can_search: " + f"{[key_emoji(key=key) for key in self.search_permissions.keys()]}" ) output += ")" return output @property def class_name(self) -> str: return str(self.__class__.__name__)
[docs] def clean_copy(self) -> "StorableObject": """ This method return a copy of self, but clean up the search_permissions and read_permissions attributes. """ return StorableObject(,, tags=self.tags, description=self.description )