Source code for steamship.plugin.outputs.raw_data_plugin_output

from __future__ import annotations

import io
from typing import Any, List, Optional, Type, Union

from pydantic import BaseModel

from steamship import Tag
from steamship.base import MimeTypes
from steamship.plugin.outputs.plugin_output import PluginOutput
from steamship.utils.binary_utils import flexi_create

[docs] class RawDataPluginOutput(PluginOutput): """Represents mime-typed raw data (or a URL pointing to raw data) that can be returned to the engine. As a few examples, you can return: - Raw text: RawDataPluginOutput(string=raw_text, MimeTypes.TXT) - Markdown text: RawDataPluginOutput(string=markdown_text, MimeTypes.MKD) - A PNG image: RawDataPluginOutput(bytes=png_bytes, MimeTypes.PNG) - A JSON-serializable Dataclass: RawDataPluginOutput(json=dataclass, MimeTypes.JSON) - Steamship Blocks: RawDataPluginOutput(json=file, MimeTypes.STEAMSHIP_BLOCK_JSON) - Data uploaded to a pre-signed URL: RawDataPluginOutput(url=presigned_url, MimeTypes.TXT) The `data` field of this object will ALWAYS be Base64 encoded by the constructor. This ensures that the object is always trivially JSON-serializable over the wire, no matter what it contains. The `mimeType` field of this object should always be filled in if known. The Steamship Engine makes use of it to proactively select defaults for handling the data returned. """ data: Optional[str] = None # Note: This is **always** Base64 encoded. mime_type: Optional[str] = None tags: Optional[List[Tag]] = None def __init__( self, base64string: str = None, string: str = None, _bytes: Union[bytes, io.BytesIO] = None, json: Any = None, mime_type: str = None, tags: Optional[List[Tag]] = None, **kwargs, ): super().__init__() self.tags = tags if base64string is not None: = base64string self.mime_type = mime_type or MimeTypes.BINARY else: # Base64-encode the data field., self.mime_type, encoding = flexi_create( base64string=base64string, string=string, json=json, _bytes=_bytes, mime_type=mime_type, force_base64=True, )
[docs] @classmethod def parse_obj(cls: Type[BaseModel], obj: Any) -> BaseModel: obj["base64string"] = obj.get("data") return super().parse_obj(obj)