prefect.orion.schemas.data

Schemas for interacting with the Orion Data API.

DataDocument pydantic-model

A data document includes an encoding string and a blob of encoded data

Subclasses can define the expected type for the blob's underlying type using the generic variable D.

For example DataDocument[str] indicates that a string should be passed when creating the document and a string will be returned when it is decoded.

Source code in prefect/orion/schemas/data.py

class DataDocument(PrefectBaseModel, Generic[D]):
    """
    A data document includes an encoding string and a blob of encoded data

    Subclasses can define the expected type for the blob's underlying type using the
    generic variable `D`.

    For example `DataDocument[str]` indicates that a string should be passed when
    creating the document and a string will be returned when it is decoded.
    """

    encoding: str
    blob: bytes

    # A cache for the decoded data, see `DataDocument.decode`
    _data: D
    __slots__ = ["_data"]

    @classmethod
    def encode(
        cls: Type["DataDocument"], encoding: str, data: D, **kwargs: Any
    ) -> "DataDocument[D]":
        """
        Create a new data document

        A serializer must be registered for the given `encoding`
        """
        # Dispatch encoding
        blob = lookup_serializer(encoding).dumps(data, **kwargs)

        inst = cls(blob=blob, encoding=encoding)
        inst._cache_data(data)
        return inst

    def decode(self) -> D:
        """
        Get the data from a data document

        A serializer must be registered for the document's encoding
        """
        if self.has_cached_data():
            return self._data

        # Dispatch decoding
        data = lookup_serializer(self.encoding).loads(self.blob)

        self._cache_data(data)
        return data

    def _cache_data(self, data) -> None:
        # Use object's setattr to avoid a pydantic 'field does not exist' error
        # See https://github.com/samuelcolvin/pydantic/issues/655
        object.__setattr__(self, "_data", data)

    def has_cached_data(self):
        return hasattr(self, "_data")

    def __str__(self) -> str:
        if self.has_cached_data():
            return repr(self._data)
        else:
            return repr(self)

    def __repr__(self) -> str:
        return f"{type(self).__name__}(encoding={self.encoding!r})"

    def __rich_repr__(self):
        yield "encoding", self.encoding

DataDocument.decode

Get the data from a data document

A serializer must be registered for the document's encoding

Source code in prefect/orion/schemas/data.py

def decode(self) -> D:
    """
    Get the data from a data document

    A serializer must be registered for the document's encoding
    """
    if self.has_cached_data():
        return self._data

    # Dispatch decoding
    data = lookup_serializer(self.encoding).loads(self.blob)

    self._cache_data(data)
    return data

DataDocument.encode classmethod

Create a new data document

A serializer must be registered for the given encoding

Source code in prefect/orion/schemas/data.py

@classmethod
def encode(
    cls: Type["DataDocument"], encoding: str, data: D, **kwargs: Any
) -> "DataDocument[D]":
    """
    Create a new data document

    A serializer must be registered for the given `encoding`
    """
    # Dispatch encoding
    blob = lookup_serializer(encoding).dumps(data, **kwargs)

    inst = cls(blob=blob, encoding=encoding)
    inst._cache_data(data)
    return inst