steamship.utils package#

Submodules#

steamship.utils.binary_utils module#

steamship.utils.binary_utils.flexi_create(base64string: Optional[str] = None, data: Optional[Any] = None, string: Optional[str] = None, json: Optional[Any] = None, _bytes: Optional[Union[bytes, _io.BytesIO]] = None, mime_type=None, force_base64=False) Tuple[Any, Union[None, str], Union[None, str]][source]#

It’s convenient for some constructors to accept a variety of input types: - data (your choice) - string - json - bytes

steamship.utils.binary_utils.guess_mime(obj: Any, provided_mime: Optional[str] = None) str[source]#
steamship.utils.binary_utils.to_b64(obj: Any) str[source]#

steamship.utils.huggingface_helper module#

This class is a helper for plugins to use models hosted on Hugging Face.

It uses asyncio parallelism to make many http requests simultaneously.

steamship.utils.huggingface_helper.get_huggingface_results(blocks: List[steamship.data.block.Block], hf_model_path: str, hf_bearer_token: str, additional_params: Optional[dict] = None, timeout_seconds: int = 30, use_gpu: bool = False) List[list][source]#

steamship.utils.kv_store module#

A simple key-value store implemented atop Files and Tags.

class steamship.utils.kv_store.KeyValueStore(client: steamship.client.steamship.Steamship, store_identifier: str = 'KeyValueStore')[source]#

Bases: object

A simple key value store implemented in Steamship.

Instances of the KeyValueStore are identified by its namespace. This store_identifier corresponds to a File that will be created with a special tag identifying it.

Entries of the KeyValueStore are saved as Tag objects with:
  • Kind = “KeyValueStore”

  • Name = the key of the (kv) pair

  • Value = a dict set to the value

Note that the value is always saved as a dict object. To save a string or int, wrap it in a dict.

WARNING:

This is essentially a clever hack atop Steamship’s tag system to provide mutable key-value storage. It is in the steamship.utils package because it’s proven useful once or twice. But in general, if you find yourself heavily relying upon it, consider reaching out to us at hello@steamship.com to let us know, and we’ll up-prioritize adding a proper key-value API.

client: steamship.client.steamship.Steamship#
delete(key: str) bool[source]#

Delete the entry represented by key

get(key: str) Optional[Dict][source]#

Get the value represented by key.

items(filter_keys: Optional[List[str]] = None) List[Tuple[str, Dict[str, Any]]][source]#

Return all key-value entries as a list of (key, value) tuples.

If filter_keys is provided, only returns keys within that list.

reset()[source]#

Delete all key-values.

set(key: str, value: Dict[str, Any])[source]#

Set the entry (key, value).

store_identifier: str#

steamship.utils.metadata module#

steamship.utils.metadata.metadata_to_str(m: Union[int, float, bool, str, List, Dict]) Optional[str][source]#
steamship.utils.metadata.str_to_metadata(s: str) Optional[Union[int, float, bool, str, List, Dict]][source]#

steamship.utils.signed_urls module#

steamship.utils.signed_urls.download_from_signed_url(url: str, to_file: Optional[pathlib.Path] = None) pathlib.Path[source]#

Downloads the Signed URL to the filename desired_filename in a temporary directory on disk.

steamship.utils.signed_urls.upload_to_signed_url(url: str, _bytes: Optional[bytes] = None, filepath: Optional[pathlib.Path] = None)[source]#

Uploads either the bytes or filepath contents to the provided Signed URL.

steamship.utils.signed_urls.url_to_bytes(url: str) bytes[source]#

Downloads the Signed URL and returns the contents as bytes.

This is a helper function to consolidate Steamship Client URL fetching to ensure a single point of handling for:
  • Error messages

  • Any required manipulations for URL signed URLs

  • Any required manipulations for localstack-based environments

Note that the base API Client does not use this method on purpose: in the event of error code, it inspects the contents of the response for a SteamshipError.

steamship.utils.signed_urls.url_to_json(url: str) any[source]#

Downloads the Signed URL and returns the contents as JSON.

steamship.utils.url module#

class steamship.utils.url.Verb(value)[source]#

Bases: str, enum.Enum

An enumeration.

GET = 'GET'#
POST = 'POST'#
steamship.utils.url.apply_localstack_url_fix(url: Optional[str]) Optional[str][source]#
steamship.utils.url.is_local(base: str) bool[source]#

Check if we are running the client locally.

steamship.utils.utils module#

steamship.utils.utils.format_uri(uri: Optional[str]) Optional[str][source]#
steamship.utils.utils.safe_get(d: Dict, key: str, default: Optional[Any] = None) Optional[Any][source]#

Safely a value from dictionairy using a specific key

steamship.utils.zip_archives module#

steamship.utils.zip_archives.unzip_folder(zip_file: pathlib.Path, into_folder: Optional[pathlib.Path]) pathlib.Path[source]#

Unzips a folder on disk, returning the path to the new folder resulting.

steamship.utils.zip_archives.zip_folder(folder: pathlib.Path, into_file: Optional[pathlib.Path]) pathlib.Path[source]#

Zips a folder on disk to a co-located zip-file of the same name.

The resulting zip file does not contain the enclosing folder name provided. It contains only the children of that folder as its root elements.

Module contents#

Collection of utility functions.