Model module#

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

neomaril_codex.model.NeomarilModel#

class neomaril_codex.model.NeomarilModel(*, model_id: str, login: str | None = None, password: str | None = None, group: str = 'datarisk', group_token: str | None = None, url: str = 'https://neomaril.staging.datarisk.net/')[source]#

Bases: BaseNeomaril

Class to manage Models deployed inside Neomaril

login#

Login for authenticating with the client. You can also use the env variable NEOMARIL_USER to set this

Type:

str

password#

Password for authenticating with the client. You can also use the env variable NEOMARIL_PASSWORD to set this

Type:

str

model_id#

Model id (hash) from the model you want to access

Type:

str

group#

Group the model is inserted. Default is ‘datarisk’ (public group)

Type:

str

group_token#

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 NEOMARIL_GROUP_TOKEN

Type:

str

url#

URL to Neomaril Server. Default value is https://neomaril.staging.datarisk.net, use it to test your deployment first before changing to production. You can also use the env variable NEOMARIL_URL to set this

Type:

str

docs#

URL for the model Swagger page

Type:

str

Raises:

Example

Getting a model, testing its health and running the prediction

from neomaril_codex.model import NeomarilModelClient
from neomaril_codex.model import NeomarilModel

client = NeomarilModelClient('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:

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:

ServerError – Model deleting 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:

str

Example

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

Generates predict code for the model to be used outside Neomaril 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 (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 beeing executed, can assume values Host or Run

  • 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:

json

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) → None[source]#

Get a execution instace for that model.

Parameters:

exec_id (str) – Execution id

Raises:

ModelError – If the user tries to get a execution from a Sync model

Example

>>> model.get_model_execution('1')
health() → str[source]#

Get the model deployment process health state.

Returns:

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

Return type:

str

Example

>>> model.health()
 'OK'
predict(*, data: dict | str | NeomarilExecution | None = None, dataset: str | NeomarilDataset = None, preprocessing: NeomarilPreprocessing | None = None, group_token: str | None = None, wait_complete: bool | None = False) → dict | NeomarilExecution[source]#

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 atribute. If Async is a string with the file path with the same filename used in the source file.

  • group_token (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 NEOMARIL_GROUP_TOKEN

  • wait_complete (bool, optional) – Boolean that informs if a model training is completed (True) or not (False). Default value is False

Raises:
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, NeomarilExecution]

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

Register the model monitoring configuration at the database

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 (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

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 NEOMARIL_GROUP_TOKEN env variable

Example

>>> model.set_token('6cb64889a45a45ea8749881e30c136df')
wait_ready()[source]#

Waits the model to be with status ‘Deployed’

Example

>>> model.wait_ready()

neomaril_codex.model.NeomarilModelClient#

class neomaril_codex.model.NeomarilModelClient(*, login: str | None = None, password: str | None = None, url: str = 'https://neomaril.staging.datarisk.net/')[source]#

Bases: BaseNeomarilClient

Class for client to access Neomaril and manage models

login#

Login for authenticating with the client. You can also use the env variable NEOMARIL_USER to set this

Type:

str

password#

Password for authenticating with the client. You can also use the env variable NEOMARIL_PASSWORD to set this

Type:

str

url#

URL to Neomaril Server. Default value is https://neomaril.staging.datarisk.net, use it to test your deployment first before changing to production. You can also use the env variable NEOMARIL_URL to set this

Type:

str

Raises:

Example

Example 1: Creation and managing a Synchronous Model

from neomaril_codex.model import NeomarilModelClient
from neomaril_codex.model import NeomarilModel

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 = NeomarilModelClient('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 neomaril_codex.model import NeomarilModelClient
from neomaril_codex.model import NeomarilModel

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 = NeomarilModelClient('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, schema: str | dict | None = None, group: str = None, extra_files: list | None = None, env: str | None = None, python_version: str = '3.8', operation='Sync', input_type: str = 'json|csv|parquet', wait_for_ready: bool = True) → NeomarilModel | str[source]#

Deploy a new model to Neomaril.

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 send as well. Mandatory for Sync models

  • group (str) – Group the model is inserted. Default to ‘datarisk’ (public group)

  • 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 enviroment. This will be encrypted in the server.

  • python_version (str, optional) – Python version for the model environment. Avaliable versions are 3.8, 3.9, 3.10. Defaults to ‘3.8’

  • operation (str) – Defines wich kind operation is beeing 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 NeomarilModel instace 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 deploy process synchronously. If its False, returns nothing after sending all the data to server and runs the deploy asynchronously

Return type:

Union[NeomarilModel, 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:

json

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 = 'datarisk', group_token: str | None = None, wait_for_ready: bool = True) → NeomarilModel[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. Default is ‘datarisk’ (public group)

  • group_token (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 NEOMARIL_GROUP_TOKEN

  • 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:
Returns:

A NeomarilModel instance with the model hash from model_id

Return type:

NeomarilModel

Example

>>> model.get_model(model_id='M9c3af308c754ee7b96b2f4a273984414d40a33be90242908f9fc4aa28ba8ec4', group='ex_group')
get_model_execution(*, model_id: str, exec_id: str, group: str | None = None) → NeomarilExecution[source]#

Get a 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:

NeomarilExecution

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 (str, optional) – Text that its expected to be on the model name. It runs similar to a LIKE query on SQL

  • state (str, optional) – Text that its expected to be on the state. It runs similar to a LIKE query on SQL

  • group (str, optional) – Text that its expected to be on the group name. It runs similar to a LIKE query on SQL

  • only_deployed (bool, optional) – If its True, filter only models ready to be used (status == “Deployed”). Defaults to False

Raises:

ServerError – Unexpected server error

Returns:

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)