steamship.invocable package#

Submodules#

steamship.invocable.config module#

class steamship.invocable.config.Config[source]#

Bases: CamelModel

Base class Steamship Package and Plugin configuration objects.

extend_with_dict(d: dict, overwrite: bool = False)[source]#

Sets the attributes on this object with provided keys and values.

extend_with_json_file(path: Path, overwrite: bool = False, fail_on_missing_file: bool = True)[source]#

Extends this config object’s values with a JSON file from disk.

This is useful for applying late-bound defaults, such as API keys added to a deployment bundle.

classmethod get_config_parameters() → Dict[str, ConfigParameter][source]#
static strip_enum(default_value)[source]#

steamship.invocable.entrypoint module#

This class is necessary to be able to please the entrypoints of both localstack and AWS.

If we set the entrypoint directly to steamship.invocable.safe_handler (imported in the init from lambda_handler), AWS is happy, but localstack is not because it tries to read steamship.invocable as a py file, not a module.

If we set the entrypoint to steamship.invocable.lambda_handler.safe_handler, Localstack is happy, but AWS is not, because it tries to read lambda_handler first, which imports things from steamship.invocable, which imports things from lambda_handler.

By adding this file which basically no-ops safe_handler into steamship.invocable.entrypoint.safe_handler, both are happy.

steamship.invocable.invocable module#

Please see https://docs.steamship.com/ for information about building a Steamship Package

class steamship.invocable.invocable.Invocable(client: Optional[Steamship] = None, config: Optional[Dict[str, Any]] = None, context: Optional[InvocationContext] = None)[source]#

Bases: ABC

A Steamship microservice.

This model.py class:

  1. Provide a pre-authenticated instance of the Steamship client

  2. Provides a Lambda handler that routes to registered functions

  3. Provides useful methods connecting functions to the router.

config: Config#
classmethod config_cls() → Type[Config][source]#

Returns the configuration object for the Invocable.

By default, Steamship packages and plugins will not take any configuration. Steamship packages and plugins may declare a configuration object which extends from Config, if needed, as follows:

class MyPackageOrPlugin:
class MyConfig(Config):

…

@classmethod def config_cls(cls):

return MyPackageOrPlugin.MyConfig

context: InvocationContext#
classmethod get_config_parameters() → Dict[str, ConfigParameter][source]#
steamship.invocable.invocable.endpoint(verb: Optional[str] = None, path: Optional[str] = None, **kwargs)[source]#

By using kwargs we can tag the function with Any parameters.

steamship.invocable.invocable.get(path: str, **kwargs)[source]#
steamship.invocable.invocable.make_registering_decorator(decorator)[source]#

Returns a copy of foreignDecorator, which is identical in every way(*), except also appends a .decorator property to the callable it spits out.

(*)We can be somewhat “hygienic”, but newDecorator still isn’t signature-preserving, i.e. you will not be able to get a runtime list of parameters. For that, you need hackish libraries…but in this case, the only argument is func, so it’s not a big issue

steamship.invocable.invocable.post(path: str, **kwargs)[source]#

steamship.invocable.invocable_request module#

class steamship.invocable.invocable_request.InvocableRequest(*, clientConfig: Configuration = None, invocation: Invocation = None, loggingConfig: LoggingConfig = None, invocationContext: InvocationContext = None)[source]#

Bases: CamelModel

A request as the Steamship Hosting Framework receives it from the Engine.

This class is different from the other Request class:

  • steamship.base.request represents a request from the Steamship Client

  • this class represents a request from the Steamship Engine to a Steamship-hosted App/Plugin

It contains both a package/plugin invocation and also the client configuration in which that invocation is intended to execute.

client_config: Configuration#
invocation: Invocation#
invocation_context: InvocationContext#
logging_config: LoggingConfig#
class steamship.invocable.invocable_request.Invocation(*, httpVerb: str = None, invocationPath: str = None, arguments: Dict[str, Any] = None, config: Dict[str, Any] = None)[source]#

Bases: CamelModel

arguments: Dict[str, Any]#
config: Dict[str, Any]#
http_verb: str#
invocation_path: str#
class steamship.invocable.invocable_request.InvocationContext(*, tenantId: str = None, userId: str = None, workspaceId: str = None, invocableHandle: str = None, invocableVersionHandle: str = None, invocableInstanceHandle: str = None, invocableType: str = None, invocableOwnerId: str = None)[source]#

Bases: CamelModel

invocable_handle: str#
invocable_instance_handle: str#
invocable_owner_id: str#
invocable_type: str#
invocable_version_handle: str#
tenant_id: str#
user_id: str#
workspace_id: str#
class steamship.invocable.invocable_request.LoggingConfig(*, loggingHost: str = None, loggingPort: str = None)[source]#

Bases: CamelModel

logging_host: str#
logging_port: str#

steamship.invocable.invocable_response module#

class steamship.invocable.invocable_response.Http(*, status: int = None, base64Wrapped: bool = None, headers: Dict[str, str] = None)[source]#

Bases: CamelModel

base64_wrapped: bool#
headers: Dict[str, str]#
status: int#
class steamship.invocable.invocable_response.InvocableResponse(status: Task = None, error: SteamshipError = None, http: Http = None, data: Any = None, string: str = None, json: Any = None, _bytes: Union[bytes, io.BytesIO] = None, mime_type=None)[source]#

Bases: GenericModel, Generic[T]

Mirrors the Response object in the Steamship server.

data: T#
static error(code: int, message: Optional[str] = None, error: Optional[SteamshipError] = None, exception: Optional[Exception] = None, prefix: Optional[str] = None) → InvocableResponse[source]#

Merges a number of error channels into one unified Response object.

Aggregates all possible messages into a single ” | “-delimeted error message.

If the final resulting error message is non-null, prefixes with the provided prefix

static from_obj(obj: Any) → InvocableResponse[source]#
http: Http#
post_update(client: Client)[source]#

Pushes this response object to the corresponding Task on the Steamship Engine.

Typically apps and plugins return their results to the Engine synchronously via HTTP. But sometimes that’s not practice – for example:

  • Microsoft’s OCR endpoint returns a Job Token that can be exchanged for updates, and eventually a result.

  • Google’s AutoML can take 20-30 minutes to train.

  • Fine-tuning BERT on ECS can take an arbitrarily long amount of time.

In these cases, it can be useful for the package/plugin to occasionally post updates to the Engine outside of the Engine’s initial synchronous request-response conversation.

set_data(data: Optional[Any] = None, string: Optional[str] = None, json: Optional[Any] = None, _bytes: Optional[Union[bytes, BytesIO]] = None, mime_type=None)[source]#
status: Task#

steamship.invocable.lambda_handler module#

steamship.invocable.lambda_handler.create_handler(invocable_cls: Type[Invocable])[source]#

Deprecated wrapper function for a Steamship invocable within an AWS Lambda function. Called by code within a plugin or package.

steamship.invocable.lambda_handler.create_safe_handler(known_invocable_for_testing: Optional[Type[Invocable]] = None)[source]#
steamship.invocable.lambda_handler.encode_exception(obj)[source]#

When logging an exception ex: logging.exception(some_error), the exception must be turned into a string so that it is accepted by elasticsearch

steamship.invocable.lambda_handler.get_class_from_module(module) → Type[Invocable][source]#
steamship.invocable.lambda_handler.handler(internal_handler, event: Dict, _: Optional[Dict] = None) → dict[source]#
steamship.invocable.lambda_handler.internal_handler(invocable_cls_func: Callable[[], Type[Invocable]], event: Dict, client: Steamship, invocation_context: InvocationContext) → InvocableResponse[source]#
steamship.invocable.lambda_handler.safe_handler(event, context=None)#
steamship.invocable.lambda_handler.safely_find_invocable_class() → Type[Invocable][source]#

Safely find the invocable class within invocable code.

steamship.invocable.package_service module#

class steamship.invocable.package_service.PackageService(client: Optional[Steamship] = None, config: Optional[Dict[str, Any]] = None, context: Optional[InvocationContext] = None)[source]#

Bases: Invocable

The Abstract Base Class of a Steamship Package.

Packages may implement whatever methods they like. To expose these methods as invocable HTTP routes, annotate the method with @get or @post and the route name.

Package implementations are effectively stateless, though they will have stateful

invoke_later(method: str, verb: Verb = Verb.POST, wait_on_tasks: List[Task] = None, arguments: Dict[str, Any] = None) → Task[Any][source]#

Schedule a method for future invocation.

Parameters

  • method (str) – The method to invoke, as registered with Steamship in the @get or @post decorator.

  • verb (Verb) – The HTTP Verb to use. Default is POST.

  • wait_on_tasks (List[Task]) – A list of Task objects (or task IDs) that should be waited upon before invocation.

  • arguments (Dict[str, Any]) – The keyword arguments of the invoked method

Returns

A Task representing the future work

Return type

Task[Any]

steamship.invocable.paramater_types module#

class steamship.invocable.paramater_types.fileurl[source]#

Bases: str

Type alias that, if used in a package method argument, will cause a file upload widget to appear.

class steamship.invocable.paramater_types.longstr[source]#

Bases: str

Long string functions mostly as a type annotation for the web.

steamship.invocable.plugin_service module#

class steamship.invocable.plugin_service.PluginService(client: Optional[Steamship] = None, config: Optional[Dict[str, Any]] = None, context: Optional[InvocationContext] = None)[source]#

Bases: Invocable, Generic[IN, OUT], ABC

The Abstract Base Class of a Steamship Plugin.

All Steamship Plugins implement the operation:

  • run(PluginRequest[T]) -> Response[U]

Many plugins are effectively stateless. This run operation defines their entire capability. Examples of such stateless plugins are: - File Import Plugin - Export Plugin

Other plugins have state but in a very controlled way: - they can be trained, - this trainable process produces a “model”, - that model acts as the state on which the run method is conditioned

This model is stored in the Steamship Workspace that owns the Plugin Instance, and access to it is provided by the hosting environment that runs the model. - TODO(ted) Document this process.

These stateful plugins are called “Trainable Plugins,” and they must implement the following additional methods:

  • get_training_parameters(PluginRequest[TrainingParameterInput]) -> Response[TrainingParameterOutput]

  • train(PluginRequest[TrainPluginInput]) -> Response[TrainPluginOutput]

abstract run(request: PluginRequest[IN]) → Union[OUT, InvocableResponse[OUT]][source]#

Runs the core operation implemented by this plugin: import, export, blockify, tag, etc.

This is the method that a Steamship Plugin implements to perform its main work.

class steamship.invocable.plugin_service.TrainablePluginService(client: Optional[Steamship] = None, config: Optional[Dict[str, Any]] = None, context: Optional[InvocationContext] = None)[source]#

Bases: PluginService, Generic[IN, OUT], ABC

abstract get_training_parameters(request: PluginRequest[TrainingParameterPluginInput]) → InvocableResponse[TrainingParameterPluginOutput][source]#

Produces the trainable parameters for this plugin.

This method is run by the Steamship Engine prior to training to fetch hyperparameters.

  • The user themselves can provide hyperparameters on the TrainingParameterPluginInput object.

  • This method then transforms those into the TrainingParameterPluginOutput object, altering the user’s values if desired.

  • The Engine then takes those TrainingParameterPluginOutput and presents them on the TrainPluginInput

abstract model_cls() → Type[TrainableModel][source]#

Returns the constructor of the TrainableModel this TrainablePluginService uses.

This is required so the run method below can load the model and provide it to the subclass implementor.

run(request: PluginRequest[IN]) → Union[OUT, InvocableResponse[OUT]][source]#

Loads the trainable model before passing the request to the run_with_model handler on the subclass.

abstract run_with_model(request: PluginRequest[IN], model: TrainableModel) → Union[OUT, InvocableResponse[OUT]][source]#

Rather than implementing run(request), a TrainablePluginService implements run_with_model(request, model)

abstract train(request: PluginRequest[TrainPluginInput], model: TrainableModel) → InvocableResponse[TrainPluginOutput][source]#

Train the model.

abstract train_status(request: PluginRequest[TrainPluginInput], model: TrainableModel) → InvocableResponse[TrainPluginOutput][source]#

Train the model.

Module contents#

class steamship.invocable.Config[source]#

Bases: CamelModel

Base class Steamship Package and Plugin configuration objects.

extend_with_dict(d: dict, overwrite: bool = False)[source]#

Sets the attributes on this object with provided keys and values.

extend_with_json_file(path: Path, overwrite: bool = False, fail_on_missing_file: bool = True)[source]#

Extends this config object’s values with a JSON file from disk.

This is useful for applying late-bound defaults, such as API keys added to a deployment bundle.

classmethod get_config_parameters() → Dict[str, ConfigParameter][source]#
static strip_enum(default_value)[source]#
class steamship.invocable.Invocable(client: Optional[Steamship] = None, config: Optional[Dict[str, Any]] = None, context: Optional[InvocationContext] = None)[source]#

Bases: ABC

A Steamship microservice.

This model.py class:

  1. Provide a pre-authenticated instance of the Steamship client

  2. Provides a Lambda handler that routes to registered functions

  3. Provides useful methods connecting functions to the router.

config: Config#
classmethod config_cls() → Type[Config][source]#

Returns the configuration object for the Invocable.

By default, Steamship packages and plugins will not take any configuration. Steamship packages and plugins may declare a configuration object which extends from Config, if needed, as follows:

class MyPackageOrPlugin:
class MyConfig(Config):

…

@classmethod def config_cls(cls):

return MyPackageOrPlugin.MyConfig

context: InvocationContext#
classmethod get_config_parameters() → Dict[str, ConfigParameter][source]#
class steamship.invocable.InvocableRequest(*, clientConfig: Configuration = None, invocation: Invocation = None, loggingConfig: LoggingConfig = None, invocationContext: InvocationContext = None)[source]#

Bases: CamelModel

A request as the Steamship Hosting Framework receives it from the Engine.

This class is different from the other Request class:

  • steamship.base.request represents a request from the Steamship Client

  • this class represents a request from the Steamship Engine to a Steamship-hosted App/Plugin

It contains both a package/plugin invocation and also the client configuration in which that invocation is intended to execute.

client_config: Configuration#
invocation: Invocation#
invocation_context: InvocationContext#
logging_config: LoggingConfig#
class steamship.invocable.InvocableResponse(status: Task = None, error: SteamshipError = None, http: Http = None, data: Any = None, string: str = None, json: Any = None, _bytes: Union[bytes, io.BytesIO] = None, mime_type=None)[source]#

Bases: GenericModel, Generic[T]

Mirrors the Response object in the Steamship server.

data: T#
static error(code: int, message: Optional[str] = None, error: Optional[SteamshipError] = None, exception: Optional[Exception] = None, prefix: Optional[str] = None) → InvocableResponse[source]#

Merges a number of error channels into one unified Response object.

Aggregates all possible messages into a single ” | “-delimeted error message.

If the final resulting error message is non-null, prefixes with the provided prefix

static from_obj(obj: Any) → InvocableResponse[source]#
http: Http#
post_update(client: Client)[source]#

Pushes this response object to the corresponding Task on the Steamship Engine.

Typically apps and plugins return their results to the Engine synchronously via HTTP. But sometimes that’s not practice – for example:

  • Microsoft’s OCR endpoint returns a Job Token that can be exchanged for updates, and eventually a result.

  • Google’s AutoML can take 20-30 minutes to train.

  • Fine-tuning BERT on ECS can take an arbitrarily long amount of time.

In these cases, it can be useful for the package/plugin to occasionally post updates to the Engine outside of the Engine’s initial synchronous request-response conversation.

set_data(data: Optional[Any] = None, string: Optional[str] = None, json: Optional[Any] = None, _bytes: Optional[Union[bytes, BytesIO]] = None, mime_type=None)[source]#
status: Task#
class steamship.invocable.Invocation(*, httpVerb: str = None, invocationPath: str = None, arguments: Dict[str, Any] = None, config: Dict[str, Any] = None)[source]#

Bases: CamelModel

arguments: Dict[str, Any]#
config: Dict[str, Any]#
http_verb: str#
invocation_path: str#
class steamship.invocable.InvocationContext(*, tenantId: str = None, userId: str = None, workspaceId: str = None, invocableHandle: str = None, invocableVersionHandle: str = None, invocableInstanceHandle: str = None, invocableType: str = None, invocableOwnerId: str = None)[source]#

Bases: CamelModel

invocable_handle: str#
invocable_instance_handle: str#
invocable_owner_id: str#
invocable_type: str#
invocable_version_handle: str#
tenant_id: str#
user_id: str#
workspace_id: str#
class steamship.invocable.LoggingConfig(*, loggingHost: str = None, loggingPort: str = None)[source]#

Bases: CamelModel

logging_host: str#
logging_port: str#
class steamship.invocable.PackageService(client: Optional[Steamship] = None, config: Optional[Dict[str, Any]] = None, context: Optional[InvocationContext] = None)[source]#

Bases: Invocable

The Abstract Base Class of a Steamship Package.

Packages may implement whatever methods they like. To expose these methods as invocable HTTP routes, annotate the method with @get or @post and the route name.

Package implementations are effectively stateless, though they will have stateful

invoke_later(method: str, verb: Verb = Verb.POST, wait_on_tasks: List[Task] = None, arguments: Dict[str, Any] = None) → Task[Any][source]#

Schedule a method for future invocation.

Parameters

  • method (str) – The method to invoke, as registered with Steamship in the @get or @post decorator.

  • verb (Verb) – The HTTP Verb to use. Default is POST.

  • wait_on_tasks (List[Task]) – A list of Task objects (or task IDs) that should be waited upon before invocation.

  • arguments (Dict[str, Any]) – The keyword arguments of the invoked method

Returns

A Task representing the future work

Return type

Task[Any]

steamship.invocable.create_handler(invocable_cls: Type[Invocable])[source]#

Deprecated wrapper function for a Steamship invocable within an AWS Lambda function. Called by code within a plugin or package.

class steamship.invocable.fileurl[source]#

Bases: str

Type alias that, if used in a package method argument, will cause a file upload widget to appear.

steamship.invocable.get(path: str, **kwargs)[source]#
class steamship.invocable.longstr[source]#

Bases: str

Long string functions mostly as a type annotation for the web.

steamship.invocable.post(path: str, **kwargs)[source]#
steamship.invocable.safe_handler(event, context=None)#