steamship package#

Subpackages#

Module contents#

class steamship.Block(*, client: Client = None, id: str = None, fileId: str = None, text: str = None, tags: Optional[List[Tag]] = [], index: Optional[int] = None, mimeType: Optional[MimeTypes] = None, url: Optional[str] = None, contentURL: Optional[str] = None, uploadType: Optional[BlockUploadType] = None)[source]#

Bases: 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.

class ListRequest(*, fileId: str = None)[source]#

Bases: Request

file_id: str#
class ListResponse(*, blocks: List[Block] = [])[source]#

Bases: Response

blocks: List[Block]#
client: Client#
content_url: Optional[str]#
static create(client: Client, file_id: str, text: Optional[str] = None, tags: Optional[List[Tag]] = None, content: Optional[Union[str, bytes]] = None, url: Optional[str] = None, mime_type: Optional[MimeTypes] = None) → Block[source]#

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.

delete() → Block[source]#
file_id: str#
static get(client: Client, _id: Optional[str] = None) → Block[source]#
id: str#
index(embedding_plugin_instance: Optional[Any] = None)[source]#

Index this block.

index_in_file: Optional[int]#
is_text()[source]#

Return whether this is a text Block.

mime_type: Optional[MimeTypes]#
classmethod parse_obj(obj: Any) → BaseModel[source]#
static query(client: Client, tag_filter_query: str) → BlockQueryResponse[source]#
raw()[source]#
tags: Optional[List[Tag]]#
text: str#
upload_type: Optional[BlockUploadType]#
url: Optional[str]#
class steamship.Configuration(config_file: Optional[Path] = None, *, apiKey: SecretStr, apiBase: HttpUrl = 'https://api.steamship.com/api/v1/', appBase: HttpUrl = 'https://steamship.run/', webBase: HttpUrl = 'https://steamship.com/', workspaceId: str = None, workspaceHandle: str = None, profile: Optional[str] = None)[source]#

Bases: CamelModel

api_base: HttpUrl#
api_key: SecretStr#
app_base: HttpUrl#
static default_config_file_has_api_key() → bool[source]#
profile: Optional[str]#
static remove_api_key_from_default_config()[source]#
web_base: HttpUrl#
workspace_handle: str#
workspace_id: str#
class steamship.DocTag(value)[source]#

Bases: str, Enum

A set of name constants for Tags with a kind of TagKind.doc; appropriate for HTML and Markdown ideas.

ARTICLE = 'article'#
BLOCK_CODE = 'block-code'#
BLOCK_QUOTE = 'block-quote'#
CAPTION = 'caption'#
CHAPTER = 'chapter'#
DIV = 'div'#
DOCUMENT = 'document'#
EMPHASIZED = 'emphasized'#
FOOTER = 'footer'#
H1 = 'h1'#
H2 = 'h2'#
H3 = 'h3'#
H4 = 'h4'#
H5 = 'h5'#
HEADER = 'header'#
IMAGE = 'image'#
LINE = 'line'#
LIST_ITEM = 'list-item'#
MAIN = 'main'#
ORDERED_LIST = 'ordered-list'#
PAGE = 'page'#
PARAGRAPH = 'paragraph'#
PRE = 'pre'#
REGION = 'region'#
SENTENCE = 'sentence'#
SPAN = 'span'#
STRONG = 'strong'#
SUBTITLE = 'subtitle'#
TELETYPE = 'teletype'#
TEXT = 'text'#
TITLE = 'title'#
TOKEN = 'token'#
UNDERLINED = 'underlined'#
UNKNOWN = 'unknown'#
UNORDERED_LIST = 'unordered-list'#
static from_html_tag(tagname: Optional[str]) → Optional[DocTag][source]#
class steamship.EmbeddingIndex(*, client: Client = None, id: str = None, handle: str = None, name: str = None, plugin: str = None, externalId: str = None, externalType: str = None, metadata: str = None)[source]#

Bases: CamelModel

A persistent, read-optimized index over embeddings.

client: Client#
static create(client: Client, handle: Optional[str] = None, name: Optional[str] = None, embedder_plugin_instance_handle: Optional[str] = None, fetch_if_exists: bool = True, external_id: Optional[str] = None, external_type: Optional[str] = None, metadata: Optional[Any] = None) → EmbeddingIndex[source]#
delete() → EmbeddingIndex[source]#
embed() → Task[IndexEmbedResponse][source]#
external_id: str#
external_type: str#
handle: str#
id: str#
insert(value: str, external_id: Optional[str] = None, external_type: Optional[str] = None, metadata: Optional[Union[int, float, bool, str, List, Dict]] = None, reindex: bool = True, allow_long_records=False) → IndexInsertResponse[source]#
insert_file(file_id: str, block_type: Optional[str] = None, external_id: Optional[str] = None, external_type: Optional[str] = None, metadata: Optional[Union[int, float, bool, str, List, Dict]] = None, reindex: bool = True) → IndexInsertResponse[source]#
insert_many(items: List[Union[EmbeddedItem, str]], reindex: bool = True, allow_long_records=False) → IndexInsertResponse[source]#
list_items(file_id: Optional[str] = None, block_id: Optional[str] = None, span_id: Optional[str] = None) → ListItemsResponse[source]#
metadata: str#
name: str#
classmethod parse_obj(obj: Any) → BaseModel[source]#
plugin: str#
search(query: Union[str, List[str]], k: int = 1, include_metadata: bool = False) → Task[QueryResults][source]#
class steamship.File(*, client: Client = None, id: str = None, handle: str = None, mimeType: MimeTypes = None, workspaceId: str = None, blocks: List[Block] = [], tags: List[Tag] = [], filename: str = None)[source]#

Bases: CamelModel

A file.

class CreateResponse(data: Any = None, string: str = None, _bytes: Union[bytes, io.BytesIO] = None, json: io.BytesIO = None, mime_type: str = None)[source]#

Bases: Response

data_: Any#
mime_type: str#
classmethod parse_obj(obj: Any) → Response[source]#
append_block(text: Optional[str] = None, tags: Optional[List[Tag]] = None, content: Optional[Union[str, bytes]] = None, url: Optional[str] = None, mime_type: Optional[MimeTypes] = None) → Block[source]#

Append a new block to this File. This is a convenience wrapper around Block.create(). You should provide only one of text, content, or url.

This is a server-side operation, saving the new Block to the file. The new block is appended to this client-side File as well for convenience.

blockify(plugin_instance: Optional[str] = None, wait_on_tasks: Optional[List[Task]] = None) → Task[source]#
blocks: List[Block]#
client: Client#
static create(client: Client, content: Optional[Union[str, bytes]] = None, mime_type: Optional[MimeTypes] = None, handle: Optional[str] = None, blocks: Optional[List[Block]] = None, tags: Optional[List[Tag]] = None) → File[source]#
static create_with_plugin(client: Client, plugin_instance: str, url: Optional[str] = None, mime_type: Optional[str] = None) → Task[File][source]#
delete() → File[source]#
filename: str#
static get(client: Client, _id: Optional[str] = None, handle: Optional[str] = None) → File[source]#
handle: str#
id: str#
index(plugin_instance: Optional[Any] = None) → EmbeddingIndex[source]#

Index every block in the file.

TODO(ted): Enable indexing the results of a tag query. TODO(ted): It’s hard to load the EmbeddingIndexPluginInstance with just a handle because of the chain of things that need to be created to it to function.

static list(client: Client) → ListFileResponse[source]#
mime_type: MimeTypes#
classmethod parse_obj(obj: Any) → BaseModel[source]#
static query(client: Client, tag_filter_query: str) → FileQueryResponse[source]#
raw()[source]#
refresh() → File[source]#
tag(plugin_instance: str = None, wait_on_tasks: List[Task] = None) → Task[TagResponse][source]#
tags: List[Tag]#
workspace_id: str#
class steamship.MimeTypes(value)[source]#

Bases: str, Enum

An enumeration.

BINARY = 'application/octet-stream'#
DOC = 'application/msword'#
DOCX = 'application/vnd.openxmlformats-officedocument.wordprocessingml.document'#
EPUB = 'application/epub+zip'#
FILE_JSON = 'fileJson'#
GIF = 'image/gif'#
HTML = 'text/html'#
JPG = 'image/jpeg'#
JSON = 'application/json'#
MKD = 'text/markdown'#
MP3 = 'audio/mp3'#
MP4_AUDIO = 'audio/mp4'#
MP4_VIDEO = 'video/mp4'#
PDF = 'application/pdf'#
PNG = 'image/png'#
PPT = 'applicatino/ms-powerpoint'#
PPTX = 'application/vnd.openxmlformats-officedocument.presentationml.presentation'#
RTF = 'application/rtf'#
STEAMSHIP_BLOCK_JSON = 'application/vnd.steamship-block.json.v1'#
TIFF = 'image/tiff'#
TXT = 'text/plain'#
UNKNOWN = 'unknown'#
WAV = 'audio/wav'#
WEBM_AUDIO = 'audio/webm'#
WEBM_VIDEO = 'video/webm'#
class steamship.Package(*, client: Client = None, id: str = None, handle: str = None, userId: str = None, profile: Optional[Manifest] = None, description: Optional[str] = None, readme: Optional[str] = None, isPublic: bool = False)[source]#

Bases: CamelModel

client: Client#
static create(client: Client, handle: Optional[str] = None, profile: Optional[Manifest] = None, is_public=False, fetch_if_exists=False) → Package[source]#
description: Optional[str]#
static get(client: Client, handle: str) → Package[source]#
handle: str#
id: str#
is_public: bool#
classmethod parse_obj(obj: Any) → BaseModel[source]#
profile: Optional[Manifest]#
readme: Optional[str]#
update(client: Client) → Package[source]#
user_id: str#
class steamship.PackageInstance(*, client: Client = None, id: str = None, handle: str = None, packageId: str = None, packageHandle: Optional[str] = None, userHandle: str = None, packageVersionId: str = None, packageVersionHandle: Optional[str] = None, userId: str = None, invocationURL: str = None, config: Dict[str, Any] = None, workspaceId: str = None, workspaceHandle: str = None)[source]#

Bases: CamelModel

client: Client#
config: Dict[str, Any]#
static create(client: Client, package_id: Optional[str] = None, package_handle: Optional[str] = None, package_version_id: Optional[str] = None, package_version_handle: Optional[str] = None, handle: Optional[str] = None, fetch_if_exists: Optional[bool] = None, config: Optional[Dict[str, Any]] = None) → PackageInstance[source]#
delete() → PackageInstance[source]#
full_url_for(path: str)[source]#
static get(client: Client, handle: str) → PackageInstance[source]#
handle: str#
id: str#
invocation_url: str#
invoke(path: str, verb: Verb = Verb.POST, timeout_s: Optional[float] = None, **kwargs)[source]#
load_missing_workspace_handle()[source]#
package_handle: Optional[str]#
package_id: str#
package_version_handle: Optional[str]#
package_version_id: str#
classmethod parse_obj(obj: Any) → BaseModel[source]#
user_handle: str#
user_id: str#
workspace_handle: str#
workspace_id: str#
class steamship.PackageVersion(*, client: Client = None, id: str = None, packageId: str = None, handle: str = None, configTemplate: Dict[str, Any] = None)[source]#

Bases: CamelModel

client: Client#
config_template: Dict[str, Any]#
static create(client: Client, package_id: Optional[str] = None, handle: Optional[str] = None, filename: Optional[str] = None, filebytes: Optional[bytes] = None, config_template: Optional[Dict[str, Any]] = None, hosting_handler: Optional[str] = None) → PackageVersion[source]#
handle: str#
id: str#
package_id: str#
classmethod parse_obj(obj: Any) → BaseModel[source]#
class steamship.PluginInstance(*, client: Client = None, id: str = None, handle: str = None, pluginId: str = None, pluginVersionId: str = None, pluginHandle: Optional[str] = None, pluginVersionHandle: Optional[str] = None, workspaceId: Optional[str] = None, userId: str = None, config: Dict[str, Any] = None, hostingType: Optional[HostingType] = None, hostingCpu: Optional[HostingCpu] = None, hostingMemory: Optional[HostingMemory] = None, hostingTimeout: Optional[HostingTimeout] = None, hostingEnvironment: Optional[HostingEnvironment] = None)[source]#

Bases: CamelModel

client: Client#
config: Dict[str, Any]#
static create(client: Client, plugin_id: Optional[str] = None, plugin_handle: Optional[str] = None, plugin_version_id: Optional[str] = None, plugin_version_handle: Optional[str] = None, handle: Optional[str] = None, fetch_if_exists: bool = True, config: Optional[Dict[str, Any]] = None) → PluginInstance[source]#

Create a plugin instance

When handle is empty the engine will automatically assign one fetch_if_exists controls whether we want to re-use an existing plugin instance or not.

delete() → PluginInstance[source]#
generate(input_file_id: Optional[str] = None, input_file_start_block_index: Optional[int] = None, input_file_end_block_index: Optional[int] = None, text: Optional[str] = None, block_query: Optional[str] = None, append_output_to_file: bool = False, output_file_id: Optional[str] = None, options: Optional[dict] = None)[source]#
static get(client: Client, handle: str) → PluginInstance[source]#
get_training_parameters(training_request: TrainingParameterPluginInput) → TrainingParameterPluginOutput[source]#
handle: str#
hosting_cpu: Optional[HostingCpu]#
hosting_environment: Optional[HostingEnvironment]#
hosting_memory: Optional[HostingMemory]#
hosting_timeout: Optional[HostingTimeout]#
hosting_type: Optional[HostingType]#
id: str#
classmethod parse_obj(obj: Any) → BaseModel[source]#
plugin_handle: Optional[str]#
plugin_id: str#
plugin_version_handle: Optional[str]#
plugin_version_id: str#
tag(doc: Union[str, File]) → Task[TagResponse][source]#
train(training_request: Optional[TrainingParameterPluginInput] = None, training_epochs: Optional[int] = None, export_query: Optional[str] = None, testing_holdout_percent: Optional[float] = None, test_split_seed: Optional[int] = None, training_params: Optional[Dict] = None, inference_params: Optional[Dict] = None) → Task[TrainPluginOutput][source]#

Train a plugin instance. Please provide either training_request OR the other parameters; passing training_request ignores all other parameters, but is kept for backwards compatibility.

user_id: str#
workspace_id: Optional[str]#
class steamship.PluginVersion(*, client: Client = None, id: str = None, pluginId: str = None, handle: str = None, hostingMemory: Optional[HostingMemory] = None, hostingTimeout: Optional[HostingTimeout] = None, hostingHandler: str = None, isPublic: bool = None, isDefault: bool = None, configTemplate: Dict[str, Any] = None)[source]#

Bases: CamelModel

client: Client#
config_template: Dict[str, Any]#
static create(client: Client, handle: str, plugin_id: Optional[str] = None, filename: Optional[str] = None, filebytes: Optional[bytes] = None, hosting_memory: Optional[HostingMemory] = None, hosting_timeout: Optional[HostingTimeout] = None, hosting_handler: Optional[str] = None, is_public: Optional[bool] = None, is_default: Optional[bool] = None, config_template: Optional[Dict[str, Any]] = None) → Task[PluginVersion][source]#
handle: str#
hosting_handler: str#
hosting_memory: Optional[HostingMemory]#
hosting_timeout: Optional[HostingTimeout]#
id: str#
is_default: bool#
is_public: bool#
static list(client: Client, plugin_id: Optional[str] = None, handle: Optional[str] = None, public: bool = True) → ListPluginVersionsResponse[source]#
classmethod parse_obj(obj: Any) → BaseModel[source]#
plugin_id: str#
class steamship.RuntimeEnvironments(value)[source]#

Bases: str, Enum

An enumeration.

LOCALHOST = 'localhost'#
REPLIT = 'replit'#
class steamship.Steamship(api_key: str = None, api_base: str = None, app_base: str = None, web_base: str = None, workspace: str = None, fail_if_workspace_exists: bool = False, profile: str = None, config_file: str = None, config: Configuration = None, trust_workspace_config: bool = False)[source]#

Bases: Client

Steamship Python Client.

get_workspace() → Workspace[source]#
static temporary_workspace(**kwargs) → Generator[Steamship, None, None][source]#

Create a client rooted in a temporary workspace that will be deleted after use.

static use(package_handle: str, instance_handle: Optional[str] = None, config: Optional[Dict[str, Any]] = None, version: Optional[str] = None, fetch_if_exists: bool = True, workspace_handle: Optional[str] = None, **kwargs) → PackageInstance[source]#

Creates/loads an instance of package package_handle.

The instance is named instance_handle and located in the Workspace named instance_handle. If no instance_handle is provided, the default is package_handle.

For example, one may write the following to always get back the same package instance, no matter how many times you run it, scoped into its own workspace:

`python instance = Steamship.use('package-handle', 'instance-handle') `

One may also write:

`python instance = Steamship.use('package-handle') # Instance will also be named `package-handle` `

If you wish to override the usage of a workspace named instance_handle, you can provide the workspace_handle parameter.

static use_plugin(plugin_handle: str, instance_handle: Optional[str] = None, config: Optional[Dict[str, Any]] = None, version: Optional[str] = None, fetch_if_exists: bool = True, workspace_handle: Optional[str] = None, **kwargs) → PluginInstance[source]#

Creates/loads an instance of plugin plugin_handle.

The instance is named instance_handle and located in the Workspace named instance_handle. If no instance_handle is provided, the default is plugin_handle.

For example, one may write the following to always get back the same plugin instance, no matter how many times you run it, scoped into its own workspace:

`python instance = Steamship.use_plugin('plugin-handle', 'instance-handle') `

One may also write:

`python instance = Steamship.use('plugin-handle') # Instance will also be named `plugin-handle` `

use_skill(skill: Skill, provider: Optional[Vendor] = None, instance_handle: Optional[str] = None, fetch_if_exists: Optional[bool] = True) → PluginInstance[source]#
exception steamship.SteamshipError(message: str = 'Undefined remote error', internal_message: Optional[str] = None, suggestion: Optional[str] = None, code: Optional[str] = None, error: Optional[Union[Exception, str]] = None)[source]#

Bases: Exception

code: str = None#
error: str = None#
static from_dict(d: Any) → SteamshipError[source]#

Last resort if subclass doesn’t override: pass through.

internal_message: str = None#
log()[source]#
message: str = None#
suggestion: str = None#
to_dict() → dict[source]#
class steamship.Tag(*, client: Client = None, id: str = None, fileId: str = None, blockId: Optional[str] = None, kind: Union[TagKind, str] = None, name: Optional[str] = None, value: Optional[Dict[Union[TagValueKey, str], Any]] = None, startIdx: Optional[int] = None, endIdx: Optional[int] = None, text: Optional[str] = None)[source]#

Bases: CamelModel

class DeleteRequest(*, id: str = None, fileId: str = None, blockId: str = None)[source]#

Bases: Request

block_id: str#
file_id: str#
id: str#
class ListRequest(*, fileId: str = None, blockId: str = None)[source]#

Bases: Request

block_id: str#
file_id: str#
class ListResponse(*, tags: List[Tag] = None)[source]#

Bases: Response

tags: List[Tag]#
block_id: Optional[str]#
client: Client#
static create(client: Client, file_id: Optional[str] = None, block_id: Optional[str] = None, kind: Optional[str] = None, name: Optional[str] = None, start_idx: Optional[int] = None, end_idx: Optional[int] = None, value: Optional[Dict[str, Any]] = None) → Tag[source]#
delete() → Tag[source]#
end_idx: Optional[int]#
file_id: str#
id: str#
index(plugin_instance: Optional[Any] = None)[source]#

Index this tag.

kind: Union[TagKind, str]#
name: Optional[str]#
static query(client: Client, tag_filter_query: str) → TagQueryResponse[source]#
start_idx: Optional[int]#
text: Optional[str]#
value: Optional[Dict[Union[TagValueKey, str], Any]]#
class steamship.Task(*, client: Client = None, taskId: str = None, userId: str = None, workspaceId: str = None, expect: Type = None, input: str = None, output: T = None, state: str = None, statusMessage: str = None, statusSuggestion: str = None, statusCode: str = None, statusCreatedOn: str = None, taskType: str = None, taskExecutor: str = None, taskCreatedOn: str = None, taskLastModifiedOn: str = None, remoteStatusInput: Optional[Dict] = None, remoteStatusOutput: Optional[Dict] = None, remoteStatusMessage: str = None, assignedWorker: str = None, startedAt: str = None, maxRetries: int = None, retries: int = None)[source]#

Bases: GenericCamelModel, Generic[T]

Encapsulates a unit of asynchronously performed work.

add_comment(external_id: Optional[str] = None, external_type: Optional[str] = None, external_group: Optional[str] = None, metadata: Optional[Any] = None) → TaskComment[source]#
as_error() → SteamshipError[source]#
assigned_worker: str#
client: Client#
expect: Type#
static get(client, _id: Optional[str] = None, handle: Optional[str] = None) → Task[source]#
input: str#
max_retries: int#
output: T#
classmethod parse_obj(obj: Any) → Task[source]#
post_update(fields: Optional[Set[str]] = None) → Task[source]#

Updates this task in the Steamship Engine.

refresh()[source]#
remote_status_input: Optional[Dict]#
remote_status_message: str#
remote_status_output: Optional[Dict]#
retries: int#
started_at: str#
state: str#
status_code: str#
status_created_on: str#
status_message: str#
status_suggestion: str#
task_created_on: str#
task_executor: str#
task_id: str#
task_last_modified_on: str#
task_type: str#
update(other: Optional[Task] = None)[source]#

Incorporates a Task into this object.

user_id: str#
wait(max_timeout_s: float = 180, retry_delay_s: float = 1, on_each_refresh: Optional[Callable[[int, float, Task], None]] = None)[source]#

Polls and blocks until the task has succeeded or failed (or timeout reached).

Parameters

  • max_timeout_s (int) – Max timeout in seconds. Default: 180s. After this timeout, an exception will be thrown.

  • retry_delay_s (float) – Delay between status checks. Default: 1s.

  • on_each_refresh (Optional[Callable[[int, float, Task], None]]) –

    Optional call back you can get after each refresh is made, including success state refreshes. The signature represents: (refresh #, total elapsed time, task)

    WARNING: Do not pass a long-running function to this variable. It will block the update polling.

workspace_id: str#
class steamship.TaskState[source]#

Bases: object

failed = 'failed'#
running = 'running'#
succeeded = 'succeeded'#
waiting = 'waiting'#
class steamship.Workspace(*, client: Client = None, id: str = None, handle: str = None)[source]#

Bases: CamelModel

class CreateRequest(*, id: Optional[str] = None, handle: Optional[str] = None, fetchIfExists: Optional[bool] = None, externalId: Optional[str] = None, externalType: Optional[str] = None, metadata: Optional[str] = None)[source]#

Bases: Request

external_id: Optional[str]#
external_type: Optional[str]#
fetch_if_exists: Optional[bool]#
handle: Optional[str]#
id: Optional[str]#
metadata: Optional[str]#
client: Client#
static create(client: Client, handle: Optional[str] = None, external_id: Optional[str] = None, external_type: Optional[str] = None, metadata: Optional[Any] = None, fetch_if_exists: bool = True) → Workspace[source]#
create_signed_url(request: Request) → Response[source]#
delete() → Workspace[source]#
static get(client: Client, id_: Optional[str] = None, handle: Optional[str] = None, fetch_if_exists: Optional[bool] = None) → Workspace[source]#
handle: str#
id: str#
static list(client: Client, t: Optional[str] = None) → ListWorkspacesResponse[source]#
classmethod parse_obj(obj: Any) → BaseModel[source]#
steamship.check_environment(env: RuntimeEnvironments, interactively_set_key: bool = True)[source]#