Base module#

Module with the base classes used at MLOps.

BaseMLOps#

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

Bases: object

Super base class to initialize other variables and URLs for other MLOps classes.

BaseMLOpsClient#

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

Bases: BaseMLOps

Base class for MLOps client side related classes. This is the class that contains some methods related to Client models administration. Mainly related to initialize environment and its variables, but also to generate groups. A group is a way to organize models clustering for different users and also to increase security. Each group has a unique token that should be used to run the models that belongs to that.

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:

NotImplementedError – When the environment is production, because it is not implemented yet

Example

In this example you can see how to create a group and after consult the list of groups that already exists.

from mlops_codex.base import BaseMLOpsClient

def start_group(password):
    client = BaseMLOpsClient(password)
    isCreated = client.create_group('ex_group', 'Group for example purpose')

    print(client.list_groups())

    return isCreated
create_group(*, name: str, description: str) → str[source]#

Create a group for multiple models of the same final client at the end if it returns TRUE, a message with the token for that group will be returned as a INFO message. You should keep this token information to be able to run the model of that group afterward.

Parameters:

  • name (str) – Name of the group. Must be 32 characters long and with no special characters (some parsing will be made)

  • description (str) – Short description of the group

Raises:
Returns:

Returns the group token

Return type:

str

list_groups() → list[source]#

List all existing groups.

Raises:

ServerError – Unexpected server error

Returns:

Return groups that exists in the database

Return type:

list

refresh_group_token(*, name: str, force: bool = False) → str[source]#

Refresh the group token. If the token it’s still valid it won’t be changed, unless you use parameter force = True. At the end a message with the token for that group will be returned as a INFO message. You should keep this new token information to be able to run the model of that group afterward.

Parameters:

  • name (str) – Name of the group to have the token refreshed

  • force (bool) – Force token expiration even if it’s still valid (this can make multiple models integrations stop working, so use with care)

Raises:
Returns:

Returns group token.

Return type:

str

Example

Suppose that you lost the token to access your group, you can create a new one forcing it with this method as at the example below.

from mlops_codex.base import BaseMLOpsClient

def update_group_token(model_client, group_name):
    model_client.refresh_group_token('ex_group', True)
    print(client.list_groups())

    return isCreated

MLOpsExecution#

class mlops_codex.base.MLOpsExecution(*, parent_id: str, exec_type: str, group: str | None = None, exec_id: str | None = None, login: str | None = None, password: str | None = None, url: str | None = None, group_token: str | None = None)[source]#

Bases: BaseMLOps

Base class for MLOps asynchronous model executions. With this class you can visualize the status of an execution and download the results after and execution has finished.

Parameters:

  • parent_id (str) – Model id (hash) from the model you want to access

  • exec_type (str) – Flag that contains which type of execution you use. It can be ‘AsyncModel’ or ‘Training’

  • group (Optional[str], optional) – Group the model is inserted

  • exec_id (Optional[str], optional) – Execution id

  • login (Optional[str], optional) – Login for authenticating with the client. You can also use the env variable MLOPS_USER to set this

  • password (Optional[str], optional) – Password for authenticating with the client. You can also use the env variable MLOPS_PASSWORD to set this

  • url (Optional[str], optional) – 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:

  • InputError – Invalid execution type

  • ModelError – If the execution id was not found or wasn’t possible to retrieve it

Example

In this example you can see how to get the status of an existing execution and download its results

from mlops_codex.base import MLOpsExecution
from mlops_codex.model import MLOpsModelClient

def get_execution_status(password, data_path):
    client = BaseMLOpsClient(password)
    model = client.create_model('Example notebook Async',
                        'score',
                        data_path+'app.py',
                        data_path+'model.pkl',
                        data_path+'requirements.txt',
                        python_version='3.9',
                        operation="Async",
                        input_type='csv'
                        )

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

    execution.get_status()

    execution.download_result()
download_result(*, path: str | None = './', filename: str | None = 'output.zip') → None[source]#

Gets the output of the execution.

Parameters:

  • path (Optional[str], optional) – Path of the result file. Default value is ‘./’

  • filename (Optional[str], optional) – Name of the result file. Default value is ‘output.zip’

Raises:

ExecutionError – Execution is unavailable or failed status.

execution_info() → None[source]#

Show the execution data in a better format

get_status() → dict[source]#

Gets the status of the related execution.

Raises:

ExecutionError – Execution unavailable

Returns:

Returns the execution status.

Return type:

dict

wait_ready() → None[source]#

Waits the execution until is no longer running

Example

>>> model.wait_ready()