Spaces:
Runtime error
Runtime error
File size: 5,282 Bytes
b115d50 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 |
from __future__ import annotations
from enum import Enum
from typing import Any, List, Optional, Type, Union
import requests
from pydantic import BaseModel, Field
from steamship import MimeTypes, SteamshipError
from steamship.base.client import Client
from steamship.base.model import CamelModel
from steamship.base.request import DeleteRequest, IdentifierRequest, Request
from steamship.base.response import Response
from steamship.data.tags.tag import Tag
class BlockQueryRequest(Request):
tag_filter_query: str
class BlockUploadType(str, Enum):
FILE = "file" # A file uploaded as bytes or a string
BLOCKS = "blocks" # Blocks are sent to create a file
URL = "url" # content will be fetched from a URL
NONE = "none" # No upload; plain text only
class Block(CamelModel):
"""A Block is a chunk of content within a File. It can be plain text content, image content,
video content, etc. If the content is not text, the text value may be the empty string
for backwards compatibility.
"""
client: Client = Field(None, exclude=True)
id: str = None
file_id: str = None
text: str = None
tags: Optional[List[Tag]] = []
index_in_file: Optional[int] = Field(alias="index")
mime_type: Optional[MimeTypes]
url: Optional[
str
] = None # Only for creation of blocks; used to fetch content from a public URL.
content_url: Optional[
str
] = None # For overriding the URL of the raw data for ephemeral blocks. Setting this will have no effect
upload_type: Optional[
BlockUploadType
] = None # for returning Blocks as the result of a generate request
class ListRequest(Request):
file_id: str = None
class ListResponse(Response):
blocks: List[Block] = []
@classmethod
def parse_obj(cls: Type[BaseModel], obj: Any) -> BaseModel:
# TODO (enias): This needs to be solved at the engine side
obj = obj["block"] if "block" in obj else obj
return super().parse_obj(obj)
@staticmethod
def get(
client: Client,
_id: str = None,
) -> Block:
return client.post(
"block/get",
IdentifierRequest(id=_id),
expect=Block,
)
@staticmethod
def create(
client: Client,
file_id: str,
text: str = None,
tags: List[Tag] = None,
content: Union[str, bytes] = None,
url: Optional[str] = None,
mime_type: Optional[MimeTypes] = None,
) -> Block:
"""
Create a new Block within a File specified by file_id.
You can create a Block in several ways:
- Providing raw text as the text parameter;
- Providing the content of the block as string or bytes;
- Providing a publicly accessible URL where the content is stored.
"""
if content is not None and url is not None:
raise SteamshipError("May provide content or URL, but not both when creating a Block")
if content is not None:
upload_type = BlockUploadType.FILE
elif url is not None:
upload_type = BlockUploadType.URL
else:
upload_type = BlockUploadType.NONE
req = {
"fileId": file_id,
"text": text,
"tags": [t.dict(by_alias=True) for t in tags] if tags else [],
"url": url,
"mimeType": mime_type,
"uploadType": upload_type,
}
file_data = (
("file-part", content, "multipart/form-data")
if upload_type == BlockUploadType.FILE
else None
)
return client.post(
"block/create",
req,
expect=Block,
file=file_data,
)
def delete(self) -> Block:
return self.client.post(
"block/delete",
DeleteRequest(id=self.id),
expect=Tag,
)
@staticmethod
def query(
client: Client,
tag_filter_query: str,
) -> BlockQueryResponse:
req = BlockQueryRequest(tag_filter_query=tag_filter_query)
res = client.post(
"block/query",
payload=req,
expect=BlockQueryResponse,
)
return res
def index(self, embedding_plugin_instance: Any = None):
"""Index this block."""
tags = [
Tag(
text=self.text,
file_id=self.file_id,
block_id=self.id,
kind="block",
start_idx=0,
end_idx=len(self.text),
)
]
return embedding_plugin_instance.insert(tags)
def raw(self):
if self.content_url is not None:
return requests.get(self.content_url).content
else:
return self.client.post(
"block/raw",
payload={
"id": self.id,
},
raw_response=True,
)
def is_text(self):
"""Return whether this is a text Block."""
return self.mime_type == MimeTypes.TXT
class BlockQueryResponse(Response):
blocks: List[Block]
Block.ListResponse.update_forward_refs()
Block.update_forward_refs()
|