Repository: SuperDuperDB/superduperdb
File: superduper/components/datatype.py (Python)

Code before:

    def __post_init__(self, db, artifacts):
        """Post-initialization hook.

        :param artifacts: The artifacts.
        """
        super().__post_init__(db, artifacts)
        if self.encodable in _ENCODABLES:
            self.encodable_cls = _ENCODABLES[self.encodable]
        else:
            import importlib

            self.encodable_cls = importlib.import_module(
                '.'.join(self.encodable.split('.')[:-1])
            ).__dict__[self.encodable.split('.')[-1]]

        self.bytes_encoding = self.bytes_encoding or CFG.bytes_encoding
        self.register_datatype(self)

    @property
    def artifact(self):
        """Check if the encodable is an artifact."""
        return self.encodable_cls.artifact

    def dict(self, metadata: bool = True, defaults: bool = True, refs: bool = False):
        """Get the dictionary representation of the object."""
        r = super().dict(metadata=metadata, defaults=defaults, refs=refs)
        if hasattr(self.bytes_encoding, 'value'):
            r['bytes_encoding'] = str(self.bytes_encoding.value)  # type: ignore[union-attr]
        return r

    def __call__(
        self, x: t.Optional[t.Any] = None, uri: t.Optional[str] = None
    ) -> '_BaseEncodable':
        """Create an instance of the encodable class.

        :param x: The optional content.
        :param uri: The optional URI.
        """
        return self.encodable_cls(datatype=self, x=x, uri=uri, db=self.db)

    @ensure_initialized
    def encode_data_with_identifier(self, item, info: t.Optional[t.Dict] = None):
        """Encode the item into bytes.

        :param item: The item to encode.
        :param info: The optional information dictionary.
        """
        info = info or {}
        data = self.encoder(item, info) if self.encoder else item
        sha1 = self.encodable_cls.get_hash(data)

Review comment (suggestion):
`datatype.decode(datatype.encode([1, ,2 ,3])) == np.array([1, 2, 3])` 

The data type has been changed here.

Code after:
        b = self.encode_data(item=item, info=info)
        if isinstance(b, bytes):
            return b, hashlib.sha1(b).hexdigest()
        else:
            return b, hashlib.sha1(str(b).encode()).hexdigest()


class NativeVector(BaseDataType):
    """Datatype for encoding vectors which are supported natively by databackend.

    :param dtype: Datatype of array to encode.
    """

    dtype: str = 'float64'

    def __post_init__(self, db, artifacts):
        self.encodable_cls = Native
        return super().__post_init__(db, artifacts)

    def encode_data(self, item, info=None):
        if isinstance(item, numpy.ndarray):
            item = item.tolist()
        return item

    def decode_data(self, item, info=None):
        return numpy.array(item).astype(self.dtype)


class DataType(BaseDataType):
    """A data type component that defines how data is encoded and decoded.

    :param encoder: A callable that converts an encodable object of this
                    encoder to bytes.
    :param decoder: A callable that converts bytes to an encodable object
                    of this encoder.
    :param info: An optional information dictionary.
    :param directory: The directory to store file types.
    :param encodable: The type of encodable object ('encodable',
                      'lazy_artifact', or 'file').
    :param bytes_encoding: The encoding type for bytes ('base64' or 'bytes').
    :param intermediate_type: Type of the intermediate data
           [IntermediateType.BYTES, IntermediateType.STRING]
    :param media_type: The media type.
    """

    encoder: t.Optional[t.Callable] = None  # not necessary if encodable is file
    decoder: t.Optional[t.Callable] = None
    info: t.Optional[t.Dict] = None  # TODO deprecate
    directory: t.Optional[str] = None  # TODO needed?
    encodable: str = 'encodable'
    bytes_encoding: t.Optional[str] = CFG.bytes_encoding