Model module#

Module with all classes and methods to manage the Machine Learning (ML) models deployed at MLOps.

MLOpsModel#

class mlops_codex.model.MLOpsModel(*, model_id: str, group: str, login: str | None = None, password: str | None = None, group_token: str | None = None, url: str | None = None)[source]#

Bases: BaseMLOps

Class to manage Models deployed inside MLOps

Parameters:

login (str) – Login for authenticating with the client. You can also use the env variable MLOPS_USER to set this
password (str) – Password for authenticating with the client. You can also use the env variable MLOPS_PASSWORD to set this
model_id (str) – Model id (hash) from the model you want to access
group (str) – Group the model is inserted.
group_token (str) – Token for executing the model (show when creating a group). It can be informed when getting the model or when running predictions, or using the env variable MLOPS_GROUP_TOKEN
url (str) – URL to MLOps Server. Default value is https://neomaril.datarisk.net/, use it to test your deployment first before changing to production. You can also use the env variable MLOPS_URL to set these

Raises:

ModelError – When the model can’t be accessed in the server
AuthenticationError – Invalid credentials

Example

Getting a model, testing its health and running the prediction

from mlops_codex.model import MLOpsModelClient
from mlops_codex.model import MLOpsModel

client = MLOpsModelClient('123456')

model = client.get_model(model_id='M9c3af308c754ee7b96b2f4a273984414d40a33be90242908f9fc4aa28ba8ec4',
                         group='ex_group')

if model.health() = 'OK':
    model.wait_ready()
    model.predict(model.schema)
else:
    model.restart_model(False)
    model.wait_ready()
    model.predict(model.schema)

delete()[source]#

Deletes the current model. IMPORTANT! For now this is irreversible, if you want to use the model again later you will need to upload again (and it will have a new ID).

Raises:

AuthenticationError – Raised if there is an authentication issue.
ServerError – Model deleting failed

Returns:

If model is at status=Deployed deletes the model and return a json with his information. If it isn’t Deployed it returns the message that the model is under another state

Return type:

str

Example

>>> model.delete()

disable()[source]#

Disables a model. It means that you won’t be able to perform some operations in the model Please, check with your team if you’re allowed to perform this operation

Raises:

AuthenticationError – Raised if there is an authentication issue.
ServerError – Model disable failed

Returns:

status=Deployed: disables the model and return a json. If it isn’t Deployed it returns the message that the model is under another state

Return type:

dict

Example

>>> model.disable()

generate_predict_code(*, language: str = 'curl') → str[source]#

Generates predict code for the model to be used outside MLOps Codex

Parameters:: language (str) – The generated code language. Supported languages are ‘curl’, ‘python’ or ‘javascript’
Raises:: InputError – Unsupported language
Returns:: The generated code.
Return type:: str

get_logs(*, start: str | None = None, end: str | None = None, routine: str | None = None, type: str | None = None)[source]#

Get the logs

Parameters:

start (Optional[str], optional) – Date to start filter. At the format aaaa-mm-dd
end (Optional[str], optional) – Date to end filter. At the format aaaa-mm-dd
routine (Optional[str], optional) – Type of routine beeing executed, can assume values Host or Run
type (Optional[str], optional) – Defines the type of the logs that are going to be filtered, can assume the values Ok, Error, Debug or Warning

Raises:

ServerError – Unexpected server error

Returns:

Logs list

Return type:

dict

Example

>>> model.get_logs(start='2023-01-31', end='2023-02-24', routine='Run', type='Error')
 {'Results':
    [{'ModelHash': 'M9c3af308c754ee7b96b2f4a273984414d40a33be90242908f9fc4aa28ba8ec4',
        'RegisteredAt': '2023-01-31T16:06:45.5955220Z',
        'OutputType': 'Error',
        'OutputData': '',
        'Routine': 'Run'}]
 }

get_model_execution(exec_id: str) → MLOpsExecution[source]#

Get an execution instace for that model.

Parameters:: exec_id (str) – Execution id
Raises:: ModelError – If the user tries to get an execution from a Sync model

Example

>>> model.get_model_execution('1')

health() → str[source]#

Get the model deployment process health state.

Raises:

AuthenticationError – Raised if there is an authentication issue.
ServerError – Raised if the server encounters an issue.
ModelError – Raised if it can not get the health of the model

Returns:

OK - if it is possible to get the health state NOK - if an exception occurs

Return type:

str

Example

>>> model.health()
 'OK'

model_info() → None[source]#: Show the model data in a better format

Runs a prediction from the current model.

Parameters:

data (Union[dict, str]) – The same data that is used in the source file. If Sync is a dict, the keys that are needed inside this dict are the ones in the schema attribute. If Async is a string with the file path with the same filename used in the source file.
group_token (Optional[str], optional) – Token for executing the model (show when creating a group). It can be informed when getting the model or when running predictions, or using the env variable MLOPS_GROUP_TOKEN
wait_complete (Optional[bool], optional) – Boolean that informs if a model training is completed (True) or not (False). Default value is False

Raises:

ModelError – Model is not available
InputError – Model requires a dataset or a data input

Returns:

The return of the scoring function in the source file for Sync models or the execution class for Async models.

Return type:

Union[dict, MLOpsExecution]

register_monitoring(*, preprocess_reference: str, shap_reference: str, configuration_file: str | dict, preprocess_file: str | None = None, requirements_file: str | None = None) → str[source]#

Parameters:

preprocess_reference (str) – Name of the preprocess reference
shap_reference (str) – Name of the preprocess function
configuration_file (str or dict) – Path of the configuration file, but it could be a dict
preprocess_file (Optional[str], optional) – Path of the preprocess script
requirements_file (str) – Path of the requirements file

Raises:

InputError – Invalid parameters for model creation

Returns:

Model id (hash)

Return type:

str

Example

>>> model.register_monitoring('parse', 'get_shap', configuration_file=PATH+'configuration.json', preprocess_file=PATH+'preprocess.py', requirements_file=PATH+'requirements.txt')

restart_model(*, wait_for_ready: bool = True)[source]#

Restart a model deployment process health state. Be sure your model is one of these states:

Deployed;
Disabled;
DisabledRecovery;
FailedRecovery.

Parameters:

wait_for_ready (bool) – If the model is being deployed, wait for it to be ready instead of failing the request. Defaults to True

Raises:

AuthenticationError – Raised if there is an authentication issue.
ServerError – Raised if the server encounters an issue.
ModelError – Raised if model could not be restarted.

Example

>>> model.restart_model()

set_token(group_token: str) → None[source]#

Saves the group token for this model instance.

Parameters:: group_token (str) – Token for executing the model (show when creating a group). You can set this using the MLOPS_GROUP_TOKEN env variable

Example

>>> model.set_token('6cb64889a45a45ea8749881e30c136df')

wait_ready()[source]#

Waits the model to be with status ‘Deployed’

Example

>>> model.wait_ready()

MLOpsModelClient#

class mlops_codex.model.MLOpsModelClient(*, login: str | None = None, password: str | None = None, url: str | None = None)[source]#

Bases: BaseMLOpsClient

Class for client to access MLOps and manage models

Parameters:

login (str) – Login for authenticating with the client. You can also use the env variable MLOPS_USER to set this
password (str) – Password for authenticating with the client. You can also use the env variable MLOPS_PASSWORD to set this
url (str) – URL to MLOps Server. Default value is https://neomaril.datarisk.net, use it to test your deployment first before changing to production. You can also use the env variable MLOPS_URL to set this

Raises:

AuthenticationError – Invalid credentials
ServerError – Server unavailable

Example

Example 1: Creation and managing a Synchronous Model

from mlops_codex.model import MLOpsModelClient
from mlops_codex.model import MLOpsModel

def new_sync_model(client, group, data_path):
    model = client.create_model('Model Example Sync',
                        'score',
                        data_path+'app.py',
                        data_path+'model.pkl',
                        data_path+'requirements.txt',
                        data_path+'schema.json',
                        group=group,
                        operation="Sync"
                        )

    model.register_monitoring('parse',
                    'get_shap',
                    configuration_file=data_path+'configuration.json',
                    preprocess_file=data_path+'preprocess.py',
                    requirements_file=data_path+'requirements.txt'
                    )

    return model.model_id

client = MLOpsModelClient('123456')
client.create_group('ex_group', 'Group for example purpose')

data_path = './samples/syncModel/'

model_id = new_sync_model(client, 'ex_group', data_path)

model_list = client.search_models()
print(model_list)

model = client.get_model(model_id, 'ex_group')

print(model.health())

model.wait_ready()
model.predict(model.schema)

print(model.get_logs(routine='Run'))

Example 2: creation and deployment of a Asynchronous Model

from mlops_codex.model import MLOpsModelClient
from mlops_codex.model import MLOpsModel

def new_async_model(client, group, data_path):
    model = client.create_model('Teste notebook Async',
                    'score',
                    data_path+'app.py',
                    data_path+'model.pkl',
                    data_path+'requirements.txt',
                    group=group,
                    python_version='3.9',
                    operation="Async",
                    input_type='csv'
                    )

    return model.model_id

def run_model(client, model_id, data_path):
    model = client.get_model(model_id, 'ex_group')

    execution = model.predict(data_path+'input.csv')

    return execution

client = MLOpsModelClient('123456')
client.create_group('ex_group', 'Group for example purpose')

data_path = './samples/asyncModel/'

model_id = new_async_model(client, 'ex_group', data_path)

execution = run_model(client, model_id, data_path)

execution.get_status()

execution.download_result()

create_model(*, model_name: str, model_reference: str, source_file: str, model_file: str, requirements_file: str, group: str, schema: str | dict | None = None, extra_files: list | None = None, env: str | None = None, python_version: str = '3.10', operation='Sync', input_type: str = 'json|csv|parquet', wait_for_ready: bool = True) → MLOpsModel | str[source]#

Deploy a new model to MLOps.

Parameters:

model_name (str) – The name of the model, in less than 32 characters
model_reference (str) – The name of the scoring function inside the source file
source_file (str) – Path of the source file. The file must have a scoring function that accepts two parameters: data (data for the request body of the model) and model_path (absolute path of where the file is located)
model_file (str) – Path of the model pkl file
requirements_file (str) – Path of the requirements file. The packages versions must be fixed eg: pandas==1.0
schema (Union[str, dict]) – Path to a JSON or XML file with a sample of the input for the entrypoint function. A dict with the sample input can be sending as well. Mandatory for Sync models
group (str) – Group the model is inserted.
extra_files (list, optional) – A optional list with additional files paths that should be uploaded. If the scoring function refer to this file they will be on the same folder as the source file
env (str, optional) – .env file to be used in your model environment. This will be encrypted in the server.
python_version (str, optional) – Python version for the model environment. Available versions are 3.8, 3.9, 3.10. Defaults to ‘3.10’
operation (str) – Defines which kind operation is being executed (Sync or Async). Default value is Sync
input_type (str) – The type of the input file that should be ‘json’, ‘csv’ or ‘parquet’
wait_for_ready (bool, optional) – Wait for model to be ready and returns a MLOpsModel instance with the new model. Defaults to True

Raises:

InputError – Some input parameters is invalid

Returns:

Returns the new model, if wait_for_ready=True runs the deployment process synchronously. If it’s False, returns nothing after sending all the data to server and runs the deployment asynchronously

Return type:

Union[MLOpsModel, str]

Example

>>> model = client.create_model('Model Example Sync', 'score',  './samples/syncModel/app.py', './samples/syncModel/'model.pkl', './samples/syncModel/requirements.txt','./samples/syncModel/schema.json', group=group, operation="Sync")

get_logs(*, model_id, start: str | None = None, end: str | None = None, routine: str | None = None, type: str | None = None)[source]#

Get the logs

Parameters:

model_id (str) – Model id (hash)
start (str, optional) – Date to start filter. At the format aaaa-mm-dd
end (str, optional) – Date to end filter. At the format aaaa-mm-dd
routine (str, optional) – Type of routine being executed, can assume values ‘Host’ (for deployment logs) or ‘Run’ (for execution logs)
type (str, optional) – Defines the type of the logs that are going to be filtered, can assume the values ‘Ok’, ‘Error’, ‘Debug’ or ‘Warning’

Raises:

ServerError – Unexpected server error

Returns:

Logs list

Return type:

dict

Example

>>> model.get_logs(routine='Run')
 {'Results':
    [{'ModelHash': 'B4c3af308c3e452e7b96b2f4a273984414d40a33be90242908f9fc4aa28ba8ec4',
        'RegisteredAt': '2023-02-03T16:06:45.5955220Z',
        'OutputType': 'Ok',
        'OutputData': '',
        'Routine': 'Run'}]
 }

get_model(*, model_id: str, group: str, group_token: str | None = None, wait_for_ready: bool | None = True) → MLOpsModel[source]#

Acess a model using its id

Parameters:

model_id (str) – Model id (hash) that needs to be acessed
group (str) – Group the model is inserted.
group_token (Optional[str], optional) – Token for executing the model (show when creating a group). It can be informed when getting the model or when running predictions, or using the env variable MLOPS_GROUP_TOKEN
wait_for_ready (Optional[bool], optional) – If the model is being deployed, wait for it to be ready instead of failing the request. Defaults to True

Raises:

ModelError – Model unavailable
ServerError – Unknown return from server

Returns:

A MLOpsModel instance with the model hash from model_id

Return type:

MLOpsModel

Example

>>> model.get_model(model_id='M9c3af308c754ee7b96b2f4a273984414d40a33be90242908f9fc4aa28ba8ec4', group='ex_group')

get_model_execution(*, model_id: str, exec_id: str, group: str | None = None) → MLOpsExecution[source]#

Get an execution instace (Async model only).

Parameters:

model_id (str) – Model id (hash)
exec_id (str) – Execution id
group (str, optional) – Group name, default value is None

Returns:

The new execution

Return type:

MLOpsExecution

Example

>>> model.get_model_execution( model_id='M9c3af308c754ee7b96b2f4a273984414d40a33be90242908f9fc4aa28ba8ec4', exec_id = '1')

search_models(*, name: str | None = None, state: str | None = None, group: str | None = None, only_deployed: bool = False) → list[source]#

Search for models using the name of the model

Parameters:

name (Optional[str], optional) – Text that it’s expected to be on the model name. It runs similar to a LIKE query on SQL
state (Optional[str], optional) – Text that it’s expected to be on the state. It runs similar to a LIKE query on SQL
group (Optional[str], optional) – Text that it’s expected to be on the group name. It runs similar to a LIKE query on SQL
only_deployed (Optional[bool], optional) – If it’s True, filter only models ready to be used (status == “Deployed”). Defaults to False

Raises:

ServerError – Unexpected server error

Returns:

A list with the models data, it can works like a filter depending on the arguments values

Return type:

list

Example

>>> client.search_models(group='ex_group', only_deployed=True)