wallaroo.client

class Client:

Client handle to a Wallaroo platform instance.

Objects of this class serve as the entrypoint to Wallaroo platform functionality.

Client( api_endpoint: Union[str, NoneType] = None, request_timeout: Union[int, NoneType] = None, auth_type: Union[str, NoneType] = None, gql_client: Union[gql.client.Client, NoneType] = None, interactive: Union[bool, NoneType] = None, time_format: str = '%Y-%d-%b %H:%M:%S')

Create a Client handle.

Parameters
  • Optional[str] api_endpoint: Host/port of the platform API endpoint. If not provided, the value of the WALLAROO_URL environment variable will be used.
  • Optional[int] request_timeout: Max timeout of web requests, in seconds
  • Optional[str] auth_type: Authentication type to use. Can be one of: "none", "sso", "user_password".
  • Optional[bool] interactive: If provided and True, some calls will print additional human information, or won't when False. If not provided, interactive defaults to True if running inside Jupyter and False otherwise.
  • str time_format: Preferred strftime format string for displaying timestamps in a human context.
@staticmethod
def get_urls( auth_type: Union[str, NoneType] = None, api_endpoint: Union[str, NoneType] = None) -> Tuple[Union[str, NoneType], Union[str, NoneType]]:

Method to calculate the auth values specified as defaults, as params or in ENV vars. Made static to be testable without reaching out to SSO, etc.

def list_tags(self) -> wallaroo.tag.Tags:

List all tags on the platform.

Returns

A list of all tags on the platform.

def list_models( self, workspace_id: Union[int, NoneType] = None, workspace_name: Union[str, NoneType] = None) -> wallaroo.model.ModelList:

List all models in the current or a specified workspace.

Parameters
  • workspace_id: Optional[int]: The workspace id to search in. If not provided, the current workspace id is used.
  • workspace_name: Optional[str]: The workspace name to search in. If not provided, the current workspace name is used.
Returns

A list of all models in the workspace.

def list_deployments(self) -> List[wallaroo.deployment.Deployment]:

List all deployments (active or not) on the platform.

Returns

A list of all deployments on the platform.

def search_pipelines( self, search_term: Union[str, NoneType] = None, deployed: Union[bool, NoneType] = None, created_start: Union[Datetime, NoneType] = None, created_end: Union[Datetime, NoneType] = None, updated_start: Union[Datetime, NoneType] = None, updated_end: Union[Datetime, NoneType] = None, workspace_id: Union[int, NoneType] = None, workspace_name: Union[str, NoneType] = None) -> wallaroo.pipeline_version.PipelineVersionList:

Search for pipelines. All parameters are optional, in which case the result is the same as list_pipelines(). All times are strings to be parsed by datetime.isoformat. Example:

 myclient.search_pipelines(created_end='2022-04-19 13:17:59+00:00', search_term="foo")
Parameters
  • str search_term: Will be matched against tags and model names. Example: "footag123".
  • bool deployed: Pipeline was deployed or not
  • str created_start: Pipeline was created at or after this time
  • str created_end: Pipeline was created at or before this time
  • str updated_start: Pipeline was updated at or before this time
  • str updated_end: Pipeline was updated at or before this time
  • int workspace_id: The workspace id to search in
  • str workspace_name: The workspace name to search in
Returns

A list of pipeline versions matching the search criteria.

def search_pipeline_versions( self, search_term: Union[str, NoneType] = None, deployed: Union[bool, NoneType] = None, created_start: Union[Datetime, NoneType] = None, created_end: Union[Datetime, NoneType] = None, updated_start: Union[Datetime, NoneType] = None, updated_end: Union[Datetime, NoneType] = None) -> wallaroo.pipeline_version.PipelineVersionList:

Search for pipeline versions. All parameters are optional. All times are strings to be parsed by datetime.isoformat. Example:

myclient.search_pipeline_versions(created_end='2022-04-19 13:17:59+00:00', search_term="foo")

Parameters
  • str search_term: Will be matched against tags and model names. Example: "footag123".
  • bool deployed: Pipeline was deployed or not
  • str created_start: Pipeline was created at or after this time
  • str created_end: Pipeline was created at or before this time
  • str updated_start: Pipeline was updated at or before this time
  • str updated_end: Pipeline was updated at or before this time
Returns

A list of pipeline versions matching the search criteria.

def search_models( self, search_term: Union[str, NoneType] = None, uploaded_time_start: Union[Datetime, NoneType] = None, uploaded_time_end: Union[Datetime, NoneType] = None, workspace_id: Union[int, NoneType] = None, workspace_name: Union[str, NoneType] = None) -> wallaroo.model_version.ModelVersionList:

Search all models you have access to.

Parameters
  • search_term: Optional[str]: Searches the following metadata: names, shas, versions, file names, and tags
  • uploaded_time_start: Optional[Datetime]: Inclusive time of upload
  • uploaded_time_end: Optional[Datetime] Inclusive time of upload
  • workspace_id: Optional[int]: The workspace id to search in
  • workspace_name: Optional[str]: The workspace name to search in
Returns

ModelVersionList

def search_model_versions( self, search_term: Union[str, NoneType] = None, uploaded_time_start: Union[Datetime, NoneType] = None, uploaded_time_end: Union[Datetime, NoneType] = None, workspace_id: Union[int, NoneType] = None, workspace_name: Union[str, NoneType] = None) -> wallaroo.model_version.ModelVersionList:

Search all model versions you have access to. Example:

client.search_model_versions(search_term="my_model")

Parameters
  • search_term: Optional[str]: Searches the following metadata: names, shas, versions, file names, and tags
  • uploaded_time_start: Optional["Datetime"]: Inclusive time of upload
  • uploaded_time_end: Optional["Datetime"]: Inclusive time of upload
  • workspace_id: Optional[int]: The workspace id to search in
  • workspace_name: Optional[str]: The workspace name to search in
Returns

ModelVersionList

def get_user_by_email(self, email: str) -> Union[wallaroo.user.User, NoneType]:

Find a user by email

def deactivate_user(self, email: str) -> None:

Deactivates an existing user of the platform

Deactivated users cannot log into the platform. Deactivated users do not count towards the number of allotted user seats from the license.

The Models and Pipelines owned by the deactivated user are not removed from the platform.

Parameters
  • str email: The email address of the user to deactivate.
Returns

None

def activate_user(self, email: str) -> None:

Activates an existing user of the platform that had been previously deactivated.

Activated users can log into the platform.

Parameters
  • str email: The email address of the user to activate.
Returns

None

def list_users(self) -> List[wallaroo.user.User]:

List of all Users on the platform

Returns

A list of all Users on the platform.

def upload_model( self, name: str, path: Union[str, pathlib.Path], framework: Union[wallaroo.framework.Framework, NoneType] = None, input_schema: Union[pyarrow.lib.Schema, NoneType] = None, output_schema: Union[pyarrow.lib.Schema, NoneType] = None, convert_wait: Union[bool, NoneType] = True, arch: Union[wallaroo.engine_config.Architecture, NoneType] = None, accel: Union[wallaroo.engine_config.Acceleration, NoneType] = None) -> wallaroo.model_version.ModelVersion:

Upload a model defined by a file as a new model variant.

Parameters
  • name: str The name of the model of which this is a variant. Names must be ASCII alpha-numeric characters or dash (-) only.
  • path: Union[str, pathlib.Path] Path of the model file to upload.
  • framework: Optional[Framework] Supported model frameworks. Use models from Framework Enum. Example: Framework.PYTORCH, Framework.TENSORFLOW
  • input_schema: Optional pa.Schema Input schema, required for flavors other than ONNX, Tensorflow, and Python
  • output_schema: Optional pa.Schema Output schema, required for flavors other than ONNX, Tensorflow, and Python
  • convert_wait: Optional bool Defaults to True. Specifies if method should return when conversion is over or not.
Returns

The created Model.

def generate_upload_model_api_command( self, base_url: str, name: str, path: Union[str, pathlib.Path], framework: Union[wallaroo.framework.Framework, NoneType] = None, input_schema: Union[pyarrow.lib.Schema, NoneType] = None, output_schema: Union[pyarrow.lib.Schema, NoneType] = None, arch: Union[wallaroo.engine_config.Architecture, NoneType] = None, accel: Union[wallaroo.engine_config.Acceleration, NoneType] = None) -> str:

Helper function to upload a large model via API in Wallaroo. It generates the equivalent CLI command to upload the model via API.

Parameters
  • base_url: str The base URL of the Wallaroo Cluster.
  • name: str The name of the model of which this is a variant. Names must be ASCII alpha-numeric characters or dash (-) only.
  • path: Union[str, pathlib.Path] Path of the model file to upload.
  • framework: Optional[Framework] Supported model frameworks. Use models from Framework Enum. Example: Framework.PYTORCH, Framework.TENSORFLOW
  • input_schema: Optional pa.Schema Input schema, required for flavors other than ONNX, Tensorflow, and Python
  • output_schema: Optional pa.Schema Output schema, required for flavors other than ONNX, Tensorflow, and Python
  • arch: Optional[Architecture] Supported architectures.
  • accel: Optional[Acceleration] Supported types of acceleration.
Returns

The upload_and_convert CLI command.

def register_model_image(self, name: str, image: str) -> wallaroo.model_version.ModelVersion:

Registers an MLFlow model as a new model.

Parameters
  • str model_name: The name of the model of which this is a variant. Names must be ASCII alpha-numeric characters or dash (-) only.
  • str image: Image name of the MLFlow model to register.
Returns

The created Model.

def get_model(self, name: str, version: Union[str, NoneType] = None):

Retrieves a model by name and optionally version from the current workspace.

Parameters
  • name: The name of the model.
  • version: The version of the model. If not provided, the latest version is returned.
Returns

The requested model. Raises: Exception: If the model with the given name does not exist. Exception: If the model with the given version does not exist.

def model_by_name( self, name: str, version: str, workspace_id: Union[int, NoneType] = None, workspace_name: Union[str, NoneType] = None) -> wallaroo.model_version.ModelVersion:

Fetch a Model by name.

Parameters
  • name: str: Name of the model, same as model_id.
  • version: str: Version string of the model
  • workspace_id: Optional[int]: The workspace id to search in
  • workspace_name: Optional[str]: The workspace name to search in
Returns

The Model with the corresponding model and variant name.

def model_version_by_name( self, model_class: str, model_name: str) -> wallaroo.model_version.ModelVersion:

Fetch a Model version by name.

Parameters
  • str model_class: Name of the model class.
  • str model_name: Name of the variant within the specified model class.
Returns

The Model with the corresponding model and variant name.

def deployment_by_name(self, deployment_name: str) -> wallaroo.deployment.Deployment:

Fetch a Deployment by name.

Parameters
  • str deployment_name: Name of the deployment.
Returns

The Deployment with the corresponding name.

def pipelines_by_name( self, pipeline_name: str, workspace_id: Union[int, NoneType] = None, workspace_name: Union[str, NoneType] = None) -> List[wallaroo.pipeline.Pipeline]:

Fetch Pipelines by name.

Parameters
  • str pipeline_name: Name of the pipeline.
  • int workspace_id: ID of the workspace. Defaults to None.
  • str workspace_name: Name of the workspace. Defaults to None.
Returns

The Pipeline with the corresponding name.

def list_pipelines( self, workspace_id: Union[int, NoneType] = None, workspace_name: Union[str, NoneType] = None) -> List[wallaroo.pipeline.Pipeline]:

List all pipelines on the platform.

Parameters
  • int workspace_id: ID of the workspace. Defaults to None.
  • str workspace_name: Name of the workspace. Defaults to None.
Returns

A list of all pipelines on the platform.

def get_pipeline( self, name: str, version: Union[str, NoneType] = None) -> wallaroo.pipeline.Pipeline:

Retrieves a pipeline by name and optional version from the current workspace.

Parameters
  • name: The name of the pipeline to retrieve.
  • version: The version of the pipeline to retrieve. Defaults to None.
Returns

Pipeline: The requested pipeline. Raises: Exception: If the pipeline with the given name is not found in the workspace. Exception: If the pipeline with the given version is not found in the workspace.

def build_pipeline(self, pipeline_name: str) -> wallaroo.pipeline.Pipeline:

Starts building a pipeline with the given pipeline_name, returning a :py:PipelineConfigBuilder:

When completed, the pipeline can be uploaded with .upload()

Parameters
  • pipeline_name string: Name of the pipeline, must be composed of ASCII alpha-numeric characters plus dash (-).
def get_logs( self, topic: str, limit: Union[int, NoneType] = None, start_datetime: Union[datetime.datetime, NoneType] = None, end_datetime: Union[datetime.datetime, NoneType] = None, dataset: Union[List[str], NoneType] = None, dataset_exclude: Union[List[str], NoneType] = None, dataset_separator: Union[str, NoneType] = None, directory: Union[str, NoneType] = None, file_prefix: Union[str, NoneType] = None, data_size_limit: Union[str, NoneType] = None, arrow: Union[bool, NoneType] = False) -> Tuple[Union[pyarrow.lib.Table, pandas.core.frame.DataFrame, wallaroo.logs.LogEntries, NoneType], Union[str, NoneType]]:

Get logs for the given topic.

Parameters
  • topic: str The topic to get logs for.
  • limit: Optional[int] The maximum number of logs to return.
  • start_datetime: Optional[datetime] The start time to get logs for.
  • end_datetime: Optional[datetime] The end time to get logs for. :param dataset: Optional[List[str]] By default this is set to ["*"] which returns, ["time", "in", "out", "anomaly"]. Other available options - ["metadata"]
  • dataset_exclude: Optional[List[str]] If set, allows user to exclude parts of dataset.
  • dataset_separator: Optional[Union[Sequence[str], str]] If set to ".", return dataset will be flattened.
  • directory: Optional[str] If set, logs will be exported to a file in the given directory.
  • file_prefix: Optional[str] Prefix to name the exported file. Required if directory is set.
  • data_size_limit: Optional[str] The maximum size of the exported data in MB. Size includes all files within the provided directory. By default, the data_size_limit will be set to 100MB.
  • arrow: Optional[bool] If set to True, return logs as an Arrow Table. Else, returns Pandas DataFrame.
Returns

Tuple[Union[pa.Table, pd.DataFrame, LogEntries], str] The logs and status.

def get_raw_logs( self, topic: str, start: Union[datetime.datetime, NoneType] = None, end: Union[datetime.datetime, NoneType] = None, limit: int = 100000, parse: bool = False, dataset: Union[List[str], NoneType] = None, dataset_exclude: Union[List[str], NoneType] = None, dataset_separator: Union[str, NoneType] = None, verbose: bool = False) -> Union[List[Dict[str, Any]], pandas.core.frame.DataFrame]:

Gets logs from Plateau for a particular time window without attempting to convert them to Inference LogEntries. Logs can be returned as strings or the json parsed into lists and dicts.

Parameters
  • topic str: The name of the topic to query
  • start Optional[datetime]: The start of the time window
  • end Optional[datetime]: The end of the time window
  • limit int: The number of records to retrieve. Note retrieving many records may be a performance bottleneck.
  • parse bool: Wether to attempt to parse the string as a json object.
  • verbose bool: Prints out info to help diagnose issues.
def get_raw_pipeline_inference_logs( self, topic: str, start: datetime.datetime, end: datetime.datetime, model_name: Union[str, NoneType] = None, limit: int = 100000, verbose: bool = False) -> List[Union[Dict[str, Any], pandas.core.frame.DataFrame]]:

Gets logs from Plateau for a particular time window and filters them for the model specified.

Parameters
  • pipeline_name str: The name/pipeline_id of the pipeline to query
  • topic str: The name of the topic to query
  • start Optional[datetime]: The start of the time window
  • end Optional[datetime]: The end of the time window
  • model_id: The name of the specific model to filter if any
  • limit int: The number of records to retrieve. Note retrieving many records may be a performance bottleneck.
  • verbose bool: Prints out info to help diagnose issues.
def get_pipeline_inference_dataframe( self, topic: str, start: datetime.datetime, end: datetime.datetime, model_name: Union[str, NoneType] = None, limit: int = 100000, verbose=False) -> pandas.core.frame.DataFrame:
def get_assay_results( self, assay_id: Union[str, int], start: datetime.datetime, end: datetime.datetime, workspace_id: Union[int, NoneType] = None, workspace_name: Union[str, NoneType] = None) -> wallaroo.assay.AssayAnalysisList:

Gets the assay results for a particular time window, parses them, and returns an List of AssayAnalysis.

Parameters
  • assay_id: int The id of the assay we are looking for.
  • start: datetime The start of the time window. If timezone info not set, uses UTC timezone by default.
  • end: datetime The end of the time window. If timezone info not set, uses UTC timezone by default.
  • workspace_id: Optional[int] The id of the workspace to retrieve the assay from.
  • workspace_name: Optional[str] The name of the workspace to retrieve the assay from.
Returns

List[AssayAnalysis] The assay results for the given time window.

def build_assay( self, *, assay_name: str, pipeline: wallaroo.pipeline.Pipeline, iopath: str, model_name: Union[str, NoneType] = None, baseline_start: Union[datetime.datetime, NoneType] = None, baseline_end: Union[datetime.datetime, NoneType] = None, baseline_data: Union[numpy.ndarray, NoneType] = None) -> wallaroo.assay_config.AssayBuilder:

Creates an AssayBuilder that can be used to configure and create Assays.

Parameters
  • assay_name: str Human friendly name for the assay
  • pipeline: Pipeline The pipeline this assay will work on
  • iopath: str The path to the input or output of the model that this assay will monitor.
  • model_name: Optional[str] The name of the model to use for the assay.
  • baseline_start: Optional[datetime] The start time for the inferences to use as the baseline
  • baseline_end: Optional[datetime] The end time of the baseline window. the baseline. Windows start immediately after the baseline window and are run at regular intervals continuously until the assay is deactivated or deleted.
  • baseline_data: Optional[np.ndarray] Use this to load existing baseline data at assay creation time.
def upload_assay(self, config: wallaroo.assay_config.AssayConfig) -> int:

Creates an assay in the database.

Parameters
  • config AssayConfig: The configuration for the assay to create.
Returns

The identifier for the assay that was created. :rtype int

def list_assays( self, workspace_id: Union[int, NoneType] = None, workspace_name: Union[str, NoneType] = None) -> List[wallaroo.assay.Assay]:

List all assays on the platform.

Parameters
  • workspace_id: Optional[int] The identifier for the workspace to retrieve the assays from.
  • workspace_name: Optional[str] The name of the workspace to retrieve the assays from.
Returns

A list of all assays on the platform, unless filtered by workspace.

def get_assay_info( self, assay_id: Union[int, str], workspace_id: Union[int, NoneType] = None, workspace_name: Union[str, NoneType] = None) -> pandas.core.frame.DataFrame:

Get information about a specific assay.

Parameters
  • assay_id: int The identifier for the assay to retrieve.
  • workspace_id: Optional[int] The identifier for the workspace to retrieve the assay from.
  • workspace_name: Optional[str] The name of the workspace to retrieve the assay from.
Returns

The assay with the given identifier

def set_assay_active(self, assay_id: Union[int, str], active: bool) -> None:

Sets the state of an assay to active or inactive.

Parameters
  • assay_id: int The id of the assay to set the active state of.
  • active: bool The active state to set the assay to. Default is True.
def create_tag(self, tag_text: str) -> wallaroo.tag.Tag:

Create a new tag with the given text.

def create_workspace(self, workspace_name: str) -> wallaroo.workspace.Workspace:

Create a new workspace with the current user as its first owner.

Parameters
  • str workspace_name: Name of the workspace, must be composed of ASCII alpha-numeric characters plus dash (-)
def list_workspaces(self) -> List[wallaroo.workspace.Workspace]:

List all workspaces on the platform which this user has permission see.

Returns

A list of all workspaces on the platform.

def get_workspace( self, name: str, create_if_not_exist: Union[bool, NoneType] = False) -> Union[wallaroo.workspace.Workspace, NoneType]:

Get a workspace by name. If the workspace does not exist, create it.

Parameters
  • name: The name of the workspace to get.
  • create_if_not_exist: If set to True, create a new workspace if workspace by given name doesn't already exist. Set to False by default.
Returns

The workspace with the given name.

def set_current_workspace( self, workspace: wallaroo.workspace.Workspace) -> wallaroo.workspace.Workspace:

Any calls involving pipelines or models will use the given workspace from then on.

def get_current_workspace(self) -> wallaroo.workspace.Workspace:

Return the current workspace. See also set_current_workspace.

def invite_user(self, email, password=None):
def get_topic_name(self, pipeline_pk_id: int) -> str:
def list_orchestrations( self, workspace_id: Union[int, NoneType] = None, workspace_name: Union[str, NoneType] = None) -> List[wallaroo.orchestration.Orchestration]:

List all Orchestrations in the current workspace.

Parameters
  • workspace_id: Optional[int] The ID of the workspace to list Orchestrations from.
  • workspace_name: Optional[str] The name of the workspace to list Orchestrations from.
Returns

A List containing all Orchestrations in the current workspace.

def upload_orchestration( self, bytes_buffer: Union[bytes, NoneType] = None, path: Union[str, NoneType] = None, name: Union[str, NoneType] = None, file_name: Union[str, NoneType] = None):

Upload a file to be packaged and used as an Orchestration.

The uploaded artifact must be a ZIP file which contains:

  • User code. If main.py exists, then that will be used as the task entrypoint. Otherwise, the first main.py found in any subdirectory will be used as the entrypoint.
  • Optional: A standard Python requirements.txt for any dependencies to be provided in the task environment. The Wallaroo SDK will already be present and should not be mentioned. Multiple requirements.txt files are not allowed.
  • Optional: Any other artifacts desired for runtime, including data or code.
Parameters
  • Optional[str] path: The path to the file on your filesystem that will be uploaded as an Orchestration.
  • Optional[bytes] bytes_buffer: The raw bytes to upload to be used Orchestration. Cannot be used with the path param.
  • Optional[str] name: An optional descriptive name for this Orchestration.
  • Optional[str] file_name: An optional filename to describe your Orchestration when using the bytes_buffer param. Ignored when path is used.
Returns

The Orchestration that was uploaded. :raises OrchestrationUploadFailed If a server-side error prevented the upload from succeeding.

def list_tasks( self, killed: bool = False, workspace_id: Union[int, NoneType] = None, workspace_name: Union[str, NoneType] = None) -> List[wallaroo.task.Task]:

List all Tasks in the current Workspace.

Parameters
  • killed: bool If set to True, list all killed tasks.
  • workspace_id: Optional[int] The ID of the workspace to list Tasks from.
  • workspace_name: Optional[str] The name of the workspace to list Tasks from.
Returns

A List containing Task objects.

def get_task_by_id(self, task_id: str):

Retrieve a Task by its ID.

Parameters
  • str task_id: The ID of the Task to retrieve.
Returns

A Task object.

def in_task(self) -> bool:

Determines if this code is inside an orchestration task.

Returns

True if running in a task.

def task_args(self) -> Dict[Any, Any]:

When running inside a task (see in_task()), obtain arguments passed to the task.

Returns

Dict of the arguments

def list_connections(self) -> wallaroo.connection.ConnectionList:

List all Connections defined in the platform.

Returns

List of Connections in the whole platform.

def get_connection(self, name=<class 'str'>) -> wallaroo.connection.Connection:

Retrieves a Connection by its name.

Returns

Connection to an external data source.

def create_connection( self, name=<class 'str'>, connection_type=<class 'str'>, details=typing.Dict[str, typing.Any]) -> wallaroo.connection.Connection:

Creates a Connection with the given name, type, and type-specific details.

Returns

Connection to an external data source.

def create_model_registry( self, name: str, token: str, url: str, workspace_id: Union[int, NoneType] = None) -> wallaroo.model_registry.ModelRegistry:

Create a Model Registry connection in this workspace that can be reused across workspaces.

Parameters
  • name str A descriptive name for this registry
  • token str A Bearer token necessary for accessing this Registry.
  • url str The root URL for this registry. It should NOT include /api/2.0/mlflow as part of it.
  • workspace_id int The ID of the workspace to attach this registry to, i.e. client.get_current_workspace().id().
Returns

A ModelRegistry object.

def list_model_registries(self, workspace_id: Union[int, NoneType] = None):
def get_email_by_id(self, id: str):
def remove_edge(self, name: str):

Remove an edge to a published pipeline.

Parameters
  • str name: The name of the edge that will be removed. This is not limited to this pipeline.
def mlops(self) -> wallaroo.wallaroo_ml_ops_api_client.client.AuthenticatedClient: