Source code for

# Plugin
# This file contains the abstractions for managing Steamship plugins.
# To see how to implement a Steamship Plugin, see in the same folder.

from __future__ import annotations

import json
from enum import Enum
from typing import Any, Dict, List, Optional, Type, Union

from pydantic import BaseModel, Field

from steamship.base.client import Client
from steamship.base.model import CamelModel
from steamship.base.request import IdentifierRequest, ListRequest, Request, SortOrder, UpdateRequest
from steamship.base.response import ListResponse
from import Manifest

from .hosting import HostingType

[docs] class CreatePluginRequest(Request): training_platform: Optional[HostingType] = None id: str = None type: str = None transport: str = None is_public: bool = None handle: str = None description: str = None metadata: str = None fetch_if_exists: bool = False
[docs] class PluginUpdateRequest(UpdateRequest): id: Optional[str] = None handle: Optional[str] = None description: Optional[str] = None profile: Optional[Manifest] = None readme: Optional[str] = None
[docs] class ListPluginsRequest(ListRequest): type: Optional[str] = None
[docs] class ListPluginsResponse(ListResponse): plugins: List[Plugin]
[docs] class PluginType(str, Enum): parser = "parser" classifier = "classifier" tagger = "tagger" embedder = "embedder" generator = "generator"
[docs] class PluginAdapterType(str, Enum): steamship_docker = "steamshipDocker" steamship_sagemaker = "steamshipSagemaker" huggingface = "huggingface" openai = "openai"
[docs] class PluginTargetType(str, Enum): FILE = "file" WORKSPACE = "workspace"
[docs] class Plugin(CamelModel): client: Client = Field(None, exclude=True) id: str = None type: str = None transport: str = None is_public: bool = None training_platform: Optional[HostingType] = None handle: str = None description: str = None metadata: str = None profile: Optional[Manifest] = None readme: Optional[str] = None user_id: Optional[str] = None
[docs] @classmethod def parse_obj(cls: Type[BaseModel], obj: Any) -> BaseModel: # TODO (enias): This needs to be solved at the engine side obj = obj["plugin"] if "plugin" in obj else obj return super().parse_obj(obj)
[docs] @staticmethod def create( client: Client, description: str, type_: str, transport: str, is_public: bool, handle: str = None, training_platform: Optional[HostingType] = None, metadata: Union[str, Dict, List] = None, fetch_if_exists: bool = False, ) -> Plugin: if isinstance(metadata, dict) or isinstance(metadata, list): metadata = json.dumps(metadata) req = CreatePluginRequest( training_platform=training_platform, type=type_, transport=transport, is_public=is_public, handle=handle, description=description, metadata=metadata, fetch_if_exists=fetch_if_exists, ) return "plugin/create", req, expect=Plugin, )
[docs] @staticmethod def list( client: Client, t: str = None, page_size: Optional[int] = None, page_token: Optional[str] = None, sort_order: Optional[SortOrder] = SortOrder.DESC, ) -> ListPluginsResponse: return "plugin/list", ListPluginsRequest( type=t, page_size=page_size, page_token=page_token, sort_order=sort_order ), expect=ListPluginsResponse, )
[docs] @staticmethod def get(client: Client, handle: str): return"plugin/get", IdentifierRequest(handle=handle), expect=Plugin)
[docs] def update(self, client: Client) -> Plugin: return "plugin/update", PluginUpdateRequest(, description=self.description, profile=self.profile, readme=self.readme ), expect=Plugin, )
[docs] def delete(self) -> Plugin: """Delete this plugin. If this plugin is public and another user has created an instance of it, this method will throw an error and the plugin will not be deleted. Deleting the plugin will delete any versions and instances of it that you have created.""" return "plugin/delete", IdentifierRequest(, expect=Plugin, )