Python SDK

Python package providing access to RIME’s backend services.

The main entry point should be through the Client. The other classes provide more modular functionality.

class rime_sdk.Client(domain: str, api_key: str = '', channel_timeout: float = 5.0, disable_tls: bool = False, ssl_config: Optional[Dict[str, str]] = None)

The Client provides an interface to RIME’s backend services for creating projects, starting stress test jobs, and querying the backend for current stress test jobs.

To initialize the Client, provide the address of your RIME instance.

Parameters:

domain – str The base domain/address of the RIME service.
api_key – str The api key providing authentication to RIME services.
channel_timeout – float The amount of time in seconds to wait for channels to become ready when opening connections to gRPC servers.

Raises:

ValueError – If a connection cannot be made to a backend service within timeout.

Example:

rime_client = Client("my_vpc.rime.com", "api-key")

create_project(name: str, description: str) → Project

Create a new RIME project in RIME’s backend.

Projects allow you to organize stress test runs as you see fit. A natural way to organize stress test runs is to create a project for each specific ML task, such as predicting whether a transaction is fraudulent.

Parameters:

name – str Name of the new project.
description – str Description of the new project.

Returns:

A Project that allows users to interact with it. Its project_id attribute can be used in start_stress_test() and list_stress_test_jobs().

Raises:

ValueError – If the request to the Upload service failed.

Example:

project = rime_client.create_project(name='foo', description='bar')

get_project(project_id: str) → Project: Get project by project ID.

delete_project(project_id: str) → None: Delete a project in RIME’s backend.

create_managed_image(name: str, requirements: List[PipRequirement], image_type: Optional[ImageType] = None, package_requirements: Optional[List[PackageRequirement]] = None) → RIMEImageBuilder

Create a new managed Docker image with the desired custom requirements to run RIME on.

These managed Docker images are managed by the RIME backend and will automatically be upgraded when you update your version of RIME. Note: Images take a few minutes to be built.

This method returns an object that can be used to track the progress of the image building job. The new custom image is only available for use in a stress test once it has status READY.

Parameters:

name – str The (unique) name of the new managed image. This acts as the unique identifier of the managed image. The call will fail if an image with the specified name already exists.
requirements – List[ManagedImage.PipRequirement] List of additional pip requirements to be installed on the managed image. A ManagedImage.PipRequirement can be created with the helper method Client.pip_requirement. The first argument is the name of the library (e.g. tensorflow or xgboost) and the second argument is a valid pip version specifier (e.g. >=0.1.2 or ==1.0.2) or an exact version<https://peps.python.org/pep-0440/> (e.g. 1.1.2) for the library.
image_type – Optional[ImageType] An enum refering to the base image type. Valid options are: ImageType.ALL, ImageType.TABULAR, ImageType.IMAGES, or ImageType.NLP When the image_type is not specified, a default image type is used.
package_requirements – Optional[List[ManagedImage.PackageRequirement]] [BETA] An optional List of additional package requirements to install on the managed image. Currently only apt package requirements are supported. A ManagedImage.PackageRequirement can be created with the helper method Client.apt_requirement. The first argument is the name of the package (e.g. texlive or vim) and the second argument is a valid apt version specifier (e.g. =0.1.2) or a bare version (e.g. 1.1.2) for the package.

Returns:

A RIMEImageBuilder object that provides an interface for monitoring the job in the backend.

Raises:

ValueError – If the request to the ImageRegistry service failed.

Example:

requirements = [
     # Fix the version of `xgboost` to `1.0.2`.
     rime_client.pip_requirement("xgboost", "==1.0.2"),
     # We do not care about the installed version of `tensorflow`.
     rime_client.pip_requirement("tensorflow")
 ]

# Start a new image building job
builder_job = rime_client.create_managed_image("xgboost102_tensorflow",
requirements, image_type=ImageType.ALL)

# Wait until the job has finished and print out status information.
# Once this prints out the `READY` status, your image is available for
# use in stress tests.
builder_job.get_status(verbose=True, wait_until_finish=True)

has_managed_image(name: str, check_status: bool = False) → bool

Check whether managed image with name exists.

Parameters:

name – str The (unique) name of the new managed image. This acts as the unique identifier of the managed image. The call will return False if no image exists with this name, True if one does.
check_status – bool Flag to determine if the image status should be checked. If this flag is set to True, the call will return True iff the image with the specified name exists AND the image is ready to be used.

Returns:

Boolean for whether managed image with this name exists.

Example:

if rime_client.has_managed_image("xgboost102_tensorflow"):
     ....

get_managed_image(name: str) → Dict

Get managed image by name.

Parameters:: name – str The (unique) name of the new managed image. This acts as the unique identifier of the managed image. The call will raise an error if no image exists with this name.
Returns:: A dictionary with information about the managed image.

Example:

image = rime_client.get_managed_image("xgboost102_tensorflow")

delete_managed_image(name: str) → None

Delete a managed Docker image.

Parameters:: name – str The (unique) name of the managed image. This acts as the unique identifier of the managed image.

static pip_requirement(name: str, version_specifier: Optional[str] = None) → PipRequirement: Construct a PipRequirement object for use in create_managed_image().

static apt_requirement(name: str, version_specifier: Optional[str] = None) → PackageRequirement: [BETA] Construct a PackageRequirement object for create_managed_image().

static pip_library_filter(name: str, fixed_version: Optional[str] = None) → PipLibraryFilter: Construct a PipLibraryFilter object for use in list_managed_images().

list_managed_images(pip_library_filters: Optional[List[PipLibraryFilter]] = None) → Iterator[Dict]

List all the managed Docker images.

This is where the true power of the managed images feature lies. You can search for images with specific pip libraries installed so that you do not have to create a new managed image every time you need to run a stress test.

Parameters:: pip_library_filters – Optional[List[ListImagesRequest.PipLibraryFilter]] Optional list of pip libraries to filter by. Construct each ListImagesRequest.PipLibraryFilter object with the pip_library_filter convenience method.
Returns:: An iterator of all the managed images
Raises:: ValueError – If the request to the ImageRegistry service failed or the list of pip library filters is improperly specified.

Example:

# Filter for an image with catboost1.0.3 and tensorflow installed.
filters = [
    rime_client.pip_library_filter("catboost", "1.0.3"),
    rime_client.pip_library_filter("tensorflow"),
]

# Query for the images.
images = rime_client.list_managed_images(
    pip_library_filters=filters)

# To get the names of an image.
[image["name"] for image in images]

list_agents() → Iterator[Dict]

List all agents with pagination.

Returns:: An iterator of all the agents.
Raises:: ValueError – If the request to the AgentManager service failed.

Example:

# Query for the images.
agents = rime_client.list_agents()

# To get the names of agents.
[agent["name"] for agent in agents]

list_projects() → Iterator[Project]

List projects in a paginated form.

Returns:: An iterator of all the projects.
Raises:: ValueError – If the request to the ProjectManager service fails.

Example:

# Query for projects.
projects = rime_client.list_projects()

start_stress_test(test_run_config: dict, project_id: Optional[str] = None, custom_image: Optional[CustomImage] = None, rime_managed_image: Optional[str] = None, ram_request_megabytes: Optional[int] = None, cpu_request_millicores: Optional[int] = None, data_type: str = 'tabular', agent_id: Optional[str] = None, data_source_name: Optional[str] = None) → Job

Start a RIME model stress test on the backend’s ModelTesting service.

Parameters:

test_run_config – dict Configuration for the test to be run, which specifies paths to the model and datasets to used for the test.
project_id – Optional[str] Identifier for the project where the resulting test run will be stored. If not specified, the results will be stored in the default project.
custom_image – Optional[CustomImage] Specification of a customized container image to use running the model test. The image must have all dependencies required by your model. The image must specify a name for the image and optional a pull secret (of type CustomImage.PullSecret) with the name of the kubernetes pull secret used to access the given image.
rime_managed_image – Optional[str] Name of a managed image to use when running the model test. The image must have all dependencies required by your model. To create new managed images with your desired dependencies, use the client’s create_managed_image() method.
ram_request_megabytes – Optional[int] Megabytes of RAM requested for the stress test job. The limit is 2x the megabytes requested.
cpu_request_millicores – Optional[int] Millicores of CPU requested for the stress test job. The limit is 2x the millicores requested.
data_type – str Type of data this firewall test is to be run on. Should be one of tabular, nlp, images. Defaults to tabular.
agent_id – Optional[str] Identifier for the agent where the stress test will be run. If not specified, the workspace’s default agent is used.
data_source_name – Optional[str] Name of the data source which is used in the arguments. Only Specify this if you are running a config that requires this source.

Returns:

A Job providing information about the model stress test job.

Raises:

ValueError – If the request to the ModelTest service failed.

Example

This example will likely not work for you because it requires permissions to a specific S3 bucket. This demonstrates how you might specify such a configuration.

config = {
    "run_name": "Titanic",
    "data_info": {
        "label_col": "Survived",
        "ref_path": "s3://rime-datasets/titanic/titanic_example.csv",
        "eval_path": "s3://rime-datasets/titanic/titanic_example.csv"
    },
    "model_info": {
        "path": "s3://rime-models/titanic_s3_test/titanic_example_model.py"
    }
}

Run the job using the specified config and the default Docker image in the RIME backend. Store the results under project ID foo. Use the RIME Managed Image tensorflow115. This assumes you have already created the Managed Image and waited for it to be ready.

job = rime_client.start_stress_test_job(
 test_run_config=config, project_id="foo",
 rime_managed_image="tensorflow115")

get_test_run(test_run_id: str) → TestRun

Get a TestRun object for interacting with the given test_run_id.

Checks to see if the test_run_id exists, then returns TestRun object.

Parameters:: test_run_id – str ID of the test run to query for
Returns:: A TestRun object corresponding to the test_run_id

list_stress_test_jobs(status_filters: Optional[List[str]] = None) → List[Job]

Query the backend for a list of jobs filtered by status.

This is a good way to recover Job objects. Note that this only returns jobs from the last two days, because the time-to-live of job objects in the backend is set at two days.

Parameters:: status_filters – Optional[List[str]] = None Filter for selecting jobs by a union of statuses. The following list enumerates all acceptable values. [‘pending’, ‘running’, ‘failed’, ‘succeeded’] If omitted, jobs will not be filtered by status.
Returns:: A list of Job objects. These are not guaranteed to be in any sorted order.
Raises:: ValueError – If the provided status_filters array has invalid values. If the request to the ModelTest service failed.

Example:

# Get all running and succeeded jobs for project 'foo'
jobs = rime_client.list_stress_test_jobs(
    status_filters=['pending', 'succeeded'],
)

get_firewall_for_project(project_id: str) → Firewall

Get the active fw for a project if it exists.

Query the backend for an active Firewall in a specified project which can be used to perform Firewall operations. If there is no active Firewall for the project, this call will error.

Parameters:: project_id – ID of the project which contains a Firewall.
Returns:: A Firewall object.
Raises:: ValueError – If the Firewall does not exist.

Example:

# Get FW in foo-project if it exists.
firewall = rime_client.get_firewall_for_project("foo-project")

upload_file(file_path: Union[Path, str], upload_path: Optional[str] = None) → str

Upload a file to make it accessible to RIME’s backend.

The uploaded file is stored with RIME’s backend in a blob store using its file name.

Parameters:

file_path – Union[Path, str] Path to the file to be uploaded to RIME’s blob store.
upload_path – Optional[str] = None Name of the directory in the blob store file system. If omitted, a unique random string will be the directory.

Returns:

A reference to the uploaded file’s location in the blob store. This reference can be used to refer to that object when writing RIME configs. Please store this reference for future access to the file.

Raises:

FileNotFoundError – If the path file_path does not exist.
IOError – If file_path is not a file.
ValueError – If the specified upload_path is an empty string. If there was an error in obtaining a blobstore location from the RIME backend or in uploading file_path to RIME’s blob store. In the scenario the file fails to upload, the incomplete file will NOT automatically be deleted.

upload_local_image_dataset_file(file_path: Union[Path, str], image_path_key: str = 'image_path', upload_path: Optional[str] = None) → Tuple[Dict, str]

Upload an image dataset file where image files are stored locally.

The image dataset file is expected to be a list of JSON dictionaries, with an image_path_key that references an image (either an absolute path or a relative path to an image file stored locally). Every image within the file is also uploaded to blob store, and the final file is also uploaded. If your image paths already reference an external blob storage, then use upload_file instead to upload the dataset file.

Parameters:

file_path – Union[Path, str] Path to the file to be uploaded to RIME’s blob store.

Returns:

A tuple containing of a (dict, string). The dict contains the updated dataset file with image paths replaced by s3 paths. The string contains a reference to the uploaded file’s location in the blob store. This reference can be used to refer to that object when writing RIME configs. Please store this reference for future access to the file.

Raises:

FileNotFoundError – If the path file_path does not exist.
IOError – If file_path is not a file.
ValueError – If there was an error in obtaining a blobstore location from the RIME backend or in uploading file_path to RIME’s blob store. In the scenario the file fails to upload, the incomplete file will NOT automatically be deleted.

upload_directory(dir_path: Union[Path, str], upload_hidden: bool = False, upload_path: Optional[str] = None) → str

Upload a model directory to make it accessible to RIME’s backend.

The uploaded directory is stored within RIME’s backend in a blob store. All files contained within dir_path and its subdirectories are uploaded according to their relative paths within dir_path. However, if upload_hidden is False, all hidden files and subdirectories beginning with a ‘.’ are not uploaded.

Parameters:

dir_path – Union[Path, str] Path to the directory to be uploaded to RIME’s blob store.
upload_hidden – bool = False Whether or not to upload hidden files or subdirectories (ie. those beginning with a ‘.’).
upload_path – Optional[str] = None Name of the directory in the blob store file system. If omitted, a unique random string will be the directory.

Returns:

A reference to the uploaded directory’s location in the blob store. This reference can be used to refer to that object when writing RIME configs. Please store this reference for future access to the directory.

Raises:

FileNotFoundError – If the directory dir_path does not exist.
IOError – If dir_path is not a directory or contains no files.
ValueError – If the specified upload_path is an empty string. If there was an error in obtaining a blobstore location from the RIME backend or in uploading dir_path to RIME’s blob store. In the scenario the directory fails to upload, files will NOT automatically be deleted.

list_uploaded_file_urls() → Iterator[str]: Return an iterator of file paths that have been uploaded.

get_job(job_id: str) → BaseJob: Get job by id.

start_file_scan(model_file_info: dict, custom_image: Optional[CustomImage] = None, rime_managed_image: Optional[str] = None, ram_request_megabytes: Optional[int] = None, cpu_request_millicores: Optional[int] = None, agent_id: Optional[str] = None) → FileScanJob

Start a RIME model stress test on the backend’s ModelTesting service.

Parameters:

model_file_info – dict Configuration for the ML file scan, which specifies the model file or repository.
custom_image – Optional[CustomImage] Specification of a customized container image to use running the model test. The image must have all dependencies required by your model. The image must specify a name for the image and optional a pull secret (of type CustomImage.PullSecret) with the name of the kubernetes pull secret used to access the given image.
rime_managed_image – Optional[str] Name of a managed image to use when running the model test. The image must have all dependencies required by your model. To create new managed images with your desired dependencies, use the client’s create_managed_image() method.
ram_request_megabytes – Optional[int] Megabytes of RAM requested for the stress test job. The limit is 2x the megabytes requested.
cpu_request_millicores – Optional[int] Millicores of CPU requested for the stress test job. The limit is 2x the millicores requested.
agent_id – Optional[str] Identifier for the agent where the file scan job will be run. If not specified, the workspace’s default agent is used.

Returns:

A Job providing information about the ML file scan job.

Raises:

ValueError – If the request to the service failed.

Example

This example shows how to scan a huggingface model file.

model_file_info = {
    "scan_type": "huggingface",
    "scan_path": "https://huggingface.co/transformers/v2.11.0",
}

Run the job using the specified config and the default Docker image in the RIME backend. Store the results under project ID foo.

job = rime_client.start_file_scan(model_file_info)

list_file_scan_results() → Iterator[dict]

Query the backend for a list of ML file scan results.

These contain the security reports for the scanned files or repositories.

Returns:: An iterator of dictionaries containing the ML file scan results.
Raises:: ValueError – If the request to the service failed.

Example: .. code-block:: python

# Get all ML file scan results results = rime_client.list_file_scan_results()

class rime_sdk.Project(backend: RIMEBackend, project_id: str)

An interface to a RIME project.

This object provides an interface for editing, updating, and deleting projects.

backend: RIMEBackend The RIME backend used to query about the status of the job.

project_id: str The identifier for the RIME project that this object monitors.

property info: ProjectInfo: Return information about this project.

get_link() → str

Get the web app URL to the project.

This link directs to your organization’s deployment of RIME. You can view more detailed information in the web app, including information on your test runs, comparisons of those results, and models that are monitored.

Note: this is a string that should be copy-pasted into a browser.

property name: str: Return the name of this project.

property description: str: Return the description of this project.

list_test_runs() → Iterator[TestRun]: List the stress test runs associated with the project.

create_firewall(name: str, bin_size: str, test_run_id: str, run_ct_schedule: bool = False, rolling_window_duration: Optional[timedelta] = None, reference_set_time_bin: Optional[Tuple[datetime, datetime]] = None, location_type: Optional[str] = None, location_info: Optional[Dict] = None, data_params: Optional[Dict] = None, rime_managed_image: Optional[str] = None) → Firewall

Create a Firewall for a given project.

Parameters:

name – str FW name.
bin_size – str Bin size. Can be year, month, week, day, hour.
test_run_id – str ID of the stress test run that firewall will be based on.
run_ct_schedule – bool Flag for ct scheduler.
rolling_window_duration – Optional[int] Time duration of rolling window of reference set if provided. The rolling window is only supported for firewall running scheduled ct. Only one of rolling_window_seconds or reference_set_time_bin may be set.
reference_set_time_bin – Optional[Tuple[datetime, datetime]] Time bin of reference set can be set for firewall running scheduled ct. Only one of rolling_window_seconds or reference_set_time_bin may be set.
location_type – Optional[str] Type of the data location that ScheduledCT will pull data from.
location_info – Optional[Dict] Information needed to access the data location provided.
data_params – Optional[Dict] Information needed to process data from the data location provided. By default, these are obtained from your reference information. Eg. pred_col, timestamp_col, label_col, etc…
rime_managed_image – Optional[str] Name of a managed image to use when running the model test. The image must have all dependencies required by your model. To create new managed images with your desired dependencies, use the client’s create_managed_image() method.

Returns:

A Firewall object.

Raises:

ValueError – If the provided values are invalid. If the request to the Firewall service failed.

Example:

# Create FW based on foo stress test in project.
firewall = project.create_firewall(
    "firewall name", "day", "foo")

create_firewall_from_components(name: str, bin_size: str, stress_test_config: Dict[str, Any], firewall_rules: List[Dict[str, Any]], threshold_infos: List[dict], run_ct_schedule: bool = False, location_type: Optional[str] = None, location_info: Optional[Dict] = None, data_params: Optional[Dict] = None, rime_managed_image: Optional[str] = None, data_type: str = 'tabular') → Firewall

Create a Firewall for a given project.

Parameters:

name – str FW name.
bin_size – str Can be year, month, week, day, hour.
stress_test_config – dict RIME Config that indicates the testing, model, and data configurations
firewall_rules – List[Dict] Firewall Rules to update the firewall with.
threshold_infos – List[Dict] Threshold info for each summary metric.
run_ct_schedule – bool Flag for ct scheduler.
location_type – Optional[str] Type of the data location that ScheduledCT will pull data from.
location_info – Optional[Dict] Information needed to access the data location provided.
data_params – Optional[Dict] Information needed to process data from the data location provided. By default, these are obtained from your reference information. Eg. pred_col, timestamp_col, label_col, etc…
rime_managed_image – Optional[str] Name of a managed image to use when running the model test. The image must have all dependencies required by your model. To create new managed images with your desired dependencies, use the client’s create_managed_image() method.
data_type – str Type of data this firewall test is to be run on. Should be one of tabular, nlp, images. Defaults to tabular.

Returns:

A Firewall object.

Raises:

ValueError – If the provided values are invalid. If the request to the Firewall service failed.

Example:

 # Create FW manually from components.
stress_test_config = {
     "data_info": {
         "pred_col": "preds",
         "label_col": "label",
         "ref_path": "s3://my-bucket/my-data.csv",
     },
     "model_info": {"path": "s3://model-test-bucket/model.py",},
     "model_task": "Binary Classification",
 }
 firewall_rules = [
     {
         "test_name": "Unseen Categorical",
         "description": "Value must be in a required set of values",
         "is_transformation": False,
         "firewall_configs": [
             {
                 "rule_info": {
                     "feature_names": ["city"],
                     "flagged_action": "FLAG",
                 }
             }
         ],
     }
 ]
 metric_thresholds = [
     {
         "direction": "below",
         "low": 0.999,
         "medium": 0.99,
         "high": 0.9,
         "metric_name": "accuracy",
     }
 ]
 firewall = project.create_firewall_from_components(
     "firewall name",
     "day",
     stress_test_config,
     firewall_rules,
     metric_thresholds,
 )

get_firewall() → Firewall

Get the active Firewall for a project if it exists.

Query the backend for an active Firewall in this project which can be used to perform Firewall operations. If there is no active Firewall for the project, this call will error.

Returns:: A Firewall object.
Raises:: ValueError – If the Firewall does not exist.

Example:

# Get FW if it exists.
firewall = project.get_firewall()

has_firewall() → bool: Check whether a project has a firewall or not.

delete_firewall() → None: Delete firewall for this project if exists.

get_notification_settings() → Dict

Get the list of notifications for the project.

Queries the backend to get a list of notifications added to the project. The notifications are grouped by the type of the notification and each type contains a list of emails and webhooks which are added to the notification setting

Returns:: A Dictionary of notification type and corresponding emails and webhooks added for that notification type.

Example:

notification_settings = project.list_notification_settings()

add_email(email: str, notif_type_str: str) → None

Add an email to the notification settings for the given notification type.

Currently, we support 3 notification types: [“Job_Action”, “Monitoring”, “Daily_Digest”]

Example

notification_settings = project.add_email("<email>", "<notification type>")

remove_email(email: str, notif_type_str: str) → None

Remove an email from notification settings for the given notification type.

Currently, we support 3 notification types: [“Job_Action”, “Monitoring”, “Daily_Digest”]

Example

notification_settings = project.remove_email("<email>", "<notification type>")

add_webhook(webhook: str, notif_type_str: str) → None

Add a webhook to the notification settings for the given notification type.

Currently, we support 3 notification types: [“Job_Action”, “Monitoring”, “Daily_Digest”]

Example

notification_settings = project.add_webhook("<webhook>", "<notification type>")

remove_webhook(webhook: str, notif_type_str: str) → None

Remove a webhook from notification settings for the given notification type.

Currently, we support 3 notification types: [“Job_Action”, “Monitoring”, “Daily_Digest”]

Example

notification_settings = project.remove_webhook("<webhook>", "<notification type>")

delete() → None: Delete project in RIME’s backend.

class rime_sdk.Job(backend: RIMEBackend, job_id: str)

This object provides an interface for monitoring a Stress Test Job in the RIME backend.

get_test_run_id() → str: Get the test run ID for a successful job.

get_test_run() → TestRun

Get the test run object.

Raises:: ValueError if the job does not have state 'SUCCEEDED.' –

class rime_sdk.TestRun(backend: RIMEBackend, test_run_id: str)

An interface for a RIME test run.

backend: RIMEBackend The RIME backend used to query about the test run.

test_run_id: str The string identifier for the successfully completed test run.

get_link() → str

Get the web app URL which points to the Firewall Continuous Tests page.

This page contains results for all test runs. To jump to the view which shows results for this specific test run, click on the corresponding time bin in the UI.

Note: this is a string that should be copy-pasted into a browser.

get_result_df() → DataFrame

Retrieve high level summary information for a complete stress test run in a single-row dataframe.

This dataframe includes information such as model metrics on the reference and evaluation datasets, overall RIME results such as severity across tests, and high level metadata such as the project ID and model task.

By concatenating these rows together, this allows you to build a table of test run results for sake of comparison. This only works on stress test jobs that have succeeded.

Note: this does not work on <0.14.0 RIME test runs.

Returns:: A pandas.DataFrame object containing the test run result. There are a lot of columns, so it is worth viewing them with the .columns method to see what they are. Generally, these columns have information about the model and datasets as well as summary statistics like the number of failing test cases or number of high severity test cases.

Example:

test_run = client.get_test_run(some_test_run_id)
test_run_result_df = test_run.get_result_df()

get_test_cases_df(show_test_case_metrics: bool = False) → DataFrame

Retrieve all the test cases for a completed stress test run in a dataframe.

This gives you the ability to perform granular queries on test cases. For example, if you only care about subset performance tests and want to see the results on each feature, you can fetch all the test cases in a dataframe, then query on that dataframe by test type. This only works on stress test jobs that have succeeded.

Note: this will not work for test runs run on RIME versions <0.14.0.

Parameters:: show_test_case_metrics – bool = False Whether to show test case specific metrics. This could result in a sparse dataframe that is returned, since test cases return different metrics. Defaults to False.
Returns:: A pandas.DataFrame object containing the test case results. Here is a selected list of columns in the output: 1. test_run_id: ID of the parent test run. 2. features: List of features that the test case ran on. 3. test_batch_type: Type of test that was run (e.g. Subset AUC, Must be Int, etc.). 4. status: Status of the test case (e.g. Pass, Fail, Skip, etc.). 5. severity: Metric that denotes the severity of the failure of the test.

Example:

# Wait until the job has finished, since this method only works on
# SUCCEEDED jobs.
job.get_status(verbose=True, wait_until_finish=True)
# Get the test run result.
test_run = job.get_test_run()
# Dump the test cases in dataframe ``df``.
df = test_run.get_test_cases_df()

get_test_batch(test_type: str) → TestBatch

Obtain the corresponding test batch.

A TestBatch object allows a user to query the results for the corresponding test. For example, the TestBatch object representing unseen_categorical allows a user to understand the results of the unseen_categorical test to varying levels of granularity.

Parameters:: test_type – str Name of the test.
Returns:: A TestBatch representing test_type.

Example:

batch = test_run.get_test_batch("unseen_categorical")

get_test_batches() → Iterator[TestBatch]

Get all test batches for a given project.

Returns:: An iterator across TestBatch objects.

class rime_sdk.TestBatch(backend: RIMEBackend, test_run_id: str, test_type: str)

An interface for a test batch in a RIME test run.

backend: RIMEBackend The RIME backend used to query about the test run.

test_run_id: str The string identifier for the successfully completed test run.

test_type: str The unique identifer for the test type e.g. unseen_categorical.

summary(show_batch_metrics: bool = False) → Series

Obtain the test batch summary as a Pandas Series.

The summary contains high level information about a test batch. For example, the name of the test batch, the category, and the severity of the test batch as a whole.

Returns:

A Pandas Series with the following columns (and optional additional columns for batch-level metrics):

test_run_id

test_type

test_name

category

duration_in_millis

severity

failing_features

description

summary_counts.total

summary_counts.pass

summary_counts.fail

summary_counts.warning

summary_counts.skip

get_test_cases_df() → DataFrame

Obtain a dataframe which delinates all test cases.

Different tests will have different columns/information. For example, some tests may have a column representing the number of failing rows.

Returns:: A Pandas Dataframe where each row represents a test case.

class rime_sdk.Firewall(backend: RIMEBackend, firewall_id: str)

Firewall object wrapper with helpful methods for working with RIME Firewall.

backend: RIMEBackend The RIME backend used to query about the status of the job.

firewall_id: str How to refer to the FW in the backend. Use this attribute to specify the Firewall for tasks in the backend.

get_config() → dict: Return the current config of the firewall.

Warning

The output of this method may be changing over time if the reference set source is set to a rolling window.

get_metric_thresholds() → DataFrame

Return the current thresholds for metrics tracked over time.

Returns:: A Pandas DataFrame where each row is a threshold for a different metric and each column is a threshold attribute.

update_metric_thresholds(metric_name: str, low: Optional[float] = None, medium: Optional[float] = None, high: Optional[float] = None, disabled: Optional[bool] = None) → UpdateFirewallResponse: Update the current threshold for a metric tracked over time.

get_scheduled_ct_info() → Tuple[bool, dict]

Return the status of scheduled CT and the location data is pulled from.

Returns:: The first value is a boolean indicating whether Scheduled CT has been activated. The second is a dictionary containing information about the data location used to run CT.
Return type:: Tuple[bool, dict]

Example:

# Understand if Scheduled CT is running and from which location data is pulled from.
is_ct_activated, location_args = firewall.get_scheduled_ct_info()

get_firewall_rules() → dict: Return the current firewall rules.

get_data_collector() → DataCollector: Get Data Collector, create if None.

delete_firewall() → None: Delete firewall.

update_firewall_stress_test_run(stress_test_run_id: str) → UpdateFirewallResponse

Update firewall with stress test run id.

Parameters:: stress_test_run_id – Stress Test Run ID to configure new firewall
Returns:: UpdateFirewallResponse
Raises:: ValueError – If the provided status_filters array has invalid values. If the request to the ModelTest service failed.

update_stress_test_config(stress_test_config: Dict[str, Any]) → UpdateFirewallResponse

Update firewall with stress test config.

Parameters:: stress_test_config – Stress Test Config to configure new firewall
Returns:: UpdateFirewallResponse
Raises:: ValueError – If the provided values are improperly formatted If the request to the ModelTest service failed.

Warning

Updating the stress test config also resets firewall thresholds and rules.

activate_ct_schedule(location_type: str, location_info: Optional[Dict] = None, data_params: Optional[Dict] = None, rolling_window_duration: Optional[timedelta] = None, reference_set_time_bin: Optional[Tuple[datetime, datetime]] = None) → UpdateFirewallResponse

Activate CT Schedule for this firewall with a given data type.

Parameters:

location_type – Type of location that ScheduledCT will pull data from.
location_info – Information needed to access the data location provided.
data_params – Optional[Dict] Information needed to process data from the data location provided. By default, these are obtained from your reference information. Eg. pred_col, timestamp_col, label_col, etc…
rolling_window_duration – Optional[timedelta] Fixed Time duration of rolling window
reference_set_time_bin – Optional[Tuple[datetime, datetime]] Start and end times of fixed time bin

Returns:

UpdateFirewallResponse

Raises:

ValueError – If the schedule has already been activated If the data_type is invalid If the request to the Firewall service failed

deactivate_ct_schedule() → UpdateFirewallResponse

Deactivate CT Schedule for this firewall.

Returns:: UpdateFirewallResponse
Raises:: ValueError – If the request to the ModelTest service failed

update_location_info(location_type: str, location_info: Optional[Dict] = None, data_params: Optional[Dict] = None, rolling_window_duration: Optional[timedelta] = None, reference_set_time_bin: Optional[Tuple[datetime, datetime]] = None) → UpdateFirewallResponse

Update the location associated with this firewall.

Parameters:

location_type – Type of location that the firewall is associated with.
location_info – Information needed to access the data location provided.
data_params – Optional[Dict] Information needed to process data from the data location provided. By default, these are obtained from your reference information. Eg. pred_col, timestamp_col, label_col, etc…
rolling_window_duration – Optional[timedelta] Fixed Time duration of rolling window
reference_set_time_bin – Optional[Tuple[datetime, datetime]] Start and end times of fixed time bin

Returns:

UpdateFirewallResponse

Raises:

ValueError – If the location_type, location_info or data_params are invalid If the request to the Firewall service failed

update_managed_image(rime_managed_image: str) → UpdateFirewallResponse

Update the managed image associated with this firewall.

rime_managed_image: str: Name of a managed image to use when running the model test. The image must have all dependencies required by your model. To create new managed images with your desired dependencies, use the client’s create_managed_image() method.

get_managed_image() → str: Return the name of the managed image associated with this firewall.

get_link() → str

Get the web app URL to the firewall.

This link directs to your organization’s deployment of RIME. You can view more detailed information about the firewall in the web app, including helpful visualizations, key insights on your model’s performance, and explanations of test results for each batch.

Note: this is a string that should be copy-pasted into a browser.

start_continuous_test(test_run_config: dict, disable_firewall_events: bool = True, override_existing_bins: bool = False, custom_image: Optional[CustomImage] = None, rime_managed_image: Optional[str] = None, ram_request_megabytes: Optional[int] = None, cpu_request_millicores: Optional[int] = None, agent_id: Optional[str] = None, data_source_name: Optional[str] = None) → ContinuousTestJob

Start a RIME model firewall test on the backend’s ModelTesting service.

This allows you to run Firewall Test job on the RIME backend. This will run firewall on a batch of tabular data.

Parameters:

test_run_config – dict Configuration for the test to be run, which specifies paths to the model and datasets to used for the test.
custom_image – Optional[CustomImage] Specification of a customized container image to use running the model test. The image must have all dependencies required by your model. The image must specify a name for the image and optional a pull secret (of type CustomImage.PullSecret) with the name of the kubernetes pull secret used to access the given image.
rime_managed_image – Optional[str] Name of a managed image to use when running the model test. The image must have all dependencies required by your model. To create new managed images with your desired dependencies, use the client’s create_managed_image() method.
ram_request_megabytes – Optional[int] Megabytes of RAM requested for the stress test job. If none specified, will default to 4000MB. The limit is 2x the megabytes requested.
cpu_request_millicores – Optional[int] Millicores of CPU requested for the stress test job. If none specified, will default to 1500mi. The limit is 2x the millicores requested.
agent_id – Optional[str] Identifier for the agent where the continuous test will be run. If not specified, the workspace’s default agent is used.
data_source_name – Optional[str] Name of the data source which is used in the arguments. Only Specify this if you are running a config that requires this source.

Returns:

A Job providing information about the model stress test job.

Raises:

ValueError – If the request to the ModelTest service failed.

Example:

# This example will likely not work for you because it requires permissions
# to a specific S3 bucket. This demonstrates how you might specify such a
# configuration.
incremental_config = {
    "eval_path": "s3://rime-datasets/
       fraud_continuous_testing/eval_2021_04_30_to_2021_05_01.csv",
    "timestamp_col": "timestamp"
}
# Run the job using the specified config and the default Docker image in
# the RIME backend. Use the RIME Managed Image "tensorflow115".
# This assumes you have already created the Managed Image and waited for it
# to be ready.
firewall = rime_client.get_firewall("foo")
job =
    firewall.run_firewall_incremental_data(
        test_run_config=incremental_config,
        rime_managed_image="tensorflow115",
        ram_request_megabytes=8000,
        cpu_request_millicores=2000)

list_test_runs() → Iterator[ContinuousTestRun]: List the continuous test runs associated with this firewall.

get_model_status(test_run_id: Optional[str] = None) → DataFrame

Return the model status of the firewall in detail.

Parameters:

test_run_id – an optional string. if not specified, will obtain the
run. (model status corresponding to the latest test) –

Returns a Pandas Dataframe with the following columns:

name: Name of the metric
value: Value of the metric
severity: Severity of the metric
disabled: Whether or not the metric is disabled

class rime_sdk.RIMEImageBuilder(backend: RIMEBackend, name: str, requirements: Optional[List[PipRequirement]] = None, package_requirements: Optional[List[PackageRequirement]] = None)

An interface to a RIME image builder.

get_status(verbose: bool = False, wait_until_finish: bool = False, poll_rate_sec: float = 5.0) → Dict

Query the ImageRegistry service for the image’s build status.

This query includes an option to wait until the image build is finished. It will either have succeeded or failed.

Parameters:

verbose – bool whether or not to print diagnostic information such as logs.
wait_until_finish – bool whether or not to block until the image is READY or FAILED.
poll_rate_sec – float the frequency with which to poll the image’s build status.

Returns:

A dictionary representing the image’s state.

class rime_sdk.DataCollector(backend: RIMEBackend, firewall_id: str)

Data Collector wrapper with helpful methods for working with the Data Collector.

backend: RIMEBackend The RIME backend used to query about the status of the insert.

firewall_id: str ID of the Firewall associated with the Data Collector.

log_datapoints(inputs: List[Dict], datapoint_ids: Optional[List[str]] = None, timestamps: Optional[List[str]] = None, preds: Optional[List[Union[Dict, List, float, int]]] = None, labels: Optional[List[Union[Dict, int, float]]] = None, query_ids: Optional[List[Union[str, float, int]]] = None, model_id: Optional[str] = None) → None

Log Datapoints in batch.

Parameters:

inputs – List[Dict] List of inputs to log to the data collector. Each input should be provided as a dictionary. The keys should be feature names, with their corresponding values.
datapoint_ids – Optional[List[str]] List of optional ids associated with each input. Each id should be unique. Generated by default if not provided.
timestamps – Optional[List[str]] List of optional timestamps associated with each input. If not provided, by default, this is the time when log_datapoints is called.
preds – Optional[List[Union[Dict, List, float, int]]] List of optional predictions associated with each input.
labels – Optional[List[Union[Dict, int, float]]] List of optional labels associated with each input.
query_ids – Optional[List[Union[str, float, int]]] List of optional query ids associated with each input. Only relevant for ranking use cases.
model_id – Optional[str] Optional id of the model associated with the inputs and predictions logged.