mlm_insights.core.post_processors package

Subpackages

Submodules

mlm_insights.core.post_processors.local_writer_post_processor module

class mlm_insights.core.post_processors.local_writer_post_processor.LocalWriterPostProcessor(file_name: str, file_location: str, test_results_file_name: str | None = None, is_critical: bool = False)

Bases: PostProcessor

Local writer Post Processor. This will store the profile in user provided local file location.
Output will be in form of Profile object in serialized format, this object can be deserialized back using profile unmarshall method.
User need to pass valid file location for this post processor

Configuration

file_name: str

  • User-provided file name for the profile which needs to be written by post processor

file_location: str

  • User-provided file location where the profile and test results JSON file needs to be written by post processor

test_results_file_name: str

  • User-provided file name for the Insights Test results JSON file which needs to be written by post processor

is_critical: bool

  • An optional boolean to mark the run as failed in case of any exception in the post processor run. By default, the flag is set to false.

Sample code

def main():
    post_processors = [LocalWriterPostProcessor(file_name = "profile_1.bin",,
                                                test_results_file_name = "test_results_1.json",
                                                file_location = "prefix")]
    runner = InsightsBuilder().             with_input_schema(input_schema).             with_data_frame(data_frame=data_frame).             with_engine(engine=EngineDetail(engine_name="native")).             with_test_config(test_config=test_config).             with_post_processors(post_processors=post_processors).             build()

For reading from file_path which is a string
    with open(file_path, "rb") as file:
        data: Profile = file.read()
    recreated_profile = Profile.unmarshall(profile_with_feature.marshall())
process(postProcessorRequest: PostProcessorRequest, **kwargs: Any) → PostProcessorResult

Fetch the Insights Profile and Insights Test results JSON (if generated) and store it in OCI Object Storage bucket.

Parameters

postProcessorRequest: PostProcessorRequest

Returns

bool : Status of storing the Profile and Test results JSON in local file location

This returns True when the Profile and Test results are successfully uploaded in local file location, and False otherwise.

mlm_insights.core.post_processors.object_storage_writer_post_processor module

class mlm_insights.core.post_processors.object_storage_writer_post_processor.ObjectStorageWriterPostProcessor(bucket_name: str, object_name: str, prefix: str, namespace: str, test_results_object_name: str | None = None, is_critical: bool = False, **kwargs: Any)

Bases: PostProcessor

A class for interacting with Oracle Cloud Infrastructure Object Storage as a post-processor, called Object Storage Writer Post Processor.
This will store the Insights Profile and Insights Test results JSON in user-provided Object storage location.
The Profile object will be persisted in serialized format, this object can be deserialized back using profile unmarshall method.
The Test results JSON object (if generated during Insights run) will be persisted as a JSON file.

Configuration

namespace: str

  • Namespace of tenancy

bucket_name: str

  • User-configured Bucket name where this profile needs to be written by post processor

object_name: str

  • User-provided object name for profile

test_results_object_name: str

  • User-provided object name for test results JSON

prefix: str

  • User-provided object prefix

storage_options’: Dict[str, Any]

  • {“config”: “~/.oci/config”} to authenticate the file systems

is_critical: bool

  • An optional boolean to mark the run as failed in case of any exception in the post processor run. By default the flag is set to false.

Sample code

def main():
    post_processors = [ObjectStorageWriterPostProcessor(object_name = "profile_1.bin", bucket_name = "bucket",
                                                        test_results_object_name = "test_results_1.json",
                                                        object_prefix = "prefix", namespace = "namespace")]
    runner = InsightsBuilder().             with_input_schema(input_schema).             with_data_frame(data_frame=data_frame).             with_engine(engine=EngineDetail(engine_name="native")).             with_test_config(test_config=test_config).             with_post_processors(post_processors=post_processors).             build()


For reading using some datasource
    data_source_args = {
        'bucket_name': bucket_name,
        'namespace': namespace,
        'object_prefix': object_prefix,
        'storage_options' : {"config": "~/.oci/config"} to authenticate the file systems
    }
    file_location = 'oci://%s@%s/%s' % (bucket_name, namespace, object_prefix)
    fs: ocifs.OCIFileSystem = ocifs.OCIFileSystem(**data_csv)
    assert fs.exists(file_location)

    profile_file = fs.oci_client.get_object(self, namespace_name, bucket_name, object_name, **kwargs)

    # read data from profile file and unmarshall it
    with open(profile_file, "rb") as file:
        data: Profile = file.read()

    recreated_profile = Profile.unmarshall(profile_with_feature.marshall())
process(postProcessorRequest: PostProcessorRequest, **kwargs: Any) → PostProcessorResult

Fetches the Profile from PostProcessorRequest and stores it in Object Storage using default or user provided authentication policy

Parameters

postProcessorRequest: PostProcessorRequest

Returns

bool : Status of storing the Profile in Object Storage

This returns True when the Profile is successfully uploaded in Object Storage and False otherwise.

mlm_insights.core.post_processors.post_processor_request module

class mlm_insights.core.post_processors.post_processor_request.PostProcessorRequest(profile: Profile, test_results: TestResults | None = None)

Bases: ABC

Class for defining Post Processor Request needed for Insights PostProcessor Module. This will have getter’s and setter’s method for Insights profile and optionally Insights Test Results

has_test_results() → bool
property profile: Profile

Get the profile which is set during initialization.

Returns

Profile

An Instance of Profile.

property test_results: TestResults | None

Get the test results which is set during initialization.

Returns

TestResults

An Instance of TestResults.

mlm_insights.core.post_processors.post_processor_result module

class mlm_insights.core.post_processors.post_processor_result.PostProcessorResult(status: PostProcessorResultStatus, message: str, is_critical: bool = False)

Bases: ABC

Class for capturing Post Processor response .This will post processor run status and message .

mlm_insights.core.post_processors.oci_monitoring_post_processor module

class mlm_insights.core.post_processors.oci_monitoring_post_processor.OCIMonitoringPostProcessor(compartment_id: str, namespace: str = 'ml_monitoring', dimensions: Dict[str, Any] = {}, is_critical: bool = True, **kwargs: Any)

Bases: PostProcessor

OCI Monitoring Post Processor. This pushes the test suite results to the OCI Monitoring Service.
Users need to pass valid CompartmentId, Namespace (Optional), and customized dimensions in the form of key-value pairs.

Attributes

compartment_idstr

The OCID of the compartment to use for metrics.

namespacestr, optional
The namespace for the OCI monitoring (default is ‘ml_monitoring’).
Avoid entering confidential information.
dimensionsDict[str, Any], optional

Additional dimensions for the metrics (default is an empty dictionary).

is_criticalbool, optional
An optional boolean to mark the run as failed in case of any exception in the post processor run.
By default the flag is set to True.
auth: Dict[str, Any], optional
An optional dictionary containing attributes required for OCI authentication.
Sample attributes are as follows: | - file_location: str, config file location | - profile_name: str, profile name
To get more context about these attributes please refer:
https://docs.oracle.com/en-us/iaas/Content/API/Concepts/sdkconfig.htm#SDK_and_CLI_Configuration_File
process(postProcessorRequest: PostProcessorRequest, **kwargs: Any) → PostProcessorResult

Processes the results and pushes them to OCI Monitoring.

Parameters

postProcessorRequestPostProcessorRequest

The request containing results to be processed.

**kwargsAny

Additional keyword arguments.

Returns

PostProcessorResult

The result of the post-processing.

Raises

PostProcessorRunException

If there is any exception during the post-processing run.