Base module#

Module with the base classes used at Neomaril.

neomaril_codex.base.BaseNeomaril#

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

Bases: object

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

neomaril_codex.base.BaseNeomarilClient#

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

Bases: BaseNeomaril

Base class for Neomaril 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.

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:

NotImplementedError – When the environment is production, becase itis 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 neomaril_codex.base import BaseNeomarilClient

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

    print(client.list_groups())

    return isCreated
create_group(*, name: str, description: str) → bool[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 afterwards.

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:

ServerError – Unexpected server error

Returns:

Returns True if the group was successfully created and False if not

Return type:

bool

list_groups() → list[source]#

List all existing groups.

Raises:

ServerError – Unexpected server error

Returns:

List with the groups that exists in the database

Return type:

list

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

Refresh the group token. If the the token its still valid it wont 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 afterwards.

Parameters:

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

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

Raises:

ServerError – Unexpected server error

Returns:

Returns True if the group was successfully created and False if not.

Return type:

bool

Example

Supose 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 neomaril_codex.base import BaseNeomarilClient

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

    return isCreated

neomaril_codex.base.NeomarilExecution#

class neomaril_codex.base.NeomarilExecution(*, 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, group_token: str | None = None)[source]#

Bases: BaseNeomaril

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

parend_id#

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

Type:

str

exec_type#

Flag that contains which type of execution you use. It can be ‘AsyncModel’ or ‘Training’

Type:

str

group#

Group the model is inserted

Type:

str, optional

exec_id#

Execution id

Type:

str, optional

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:

  • InputError – Invalid execution type

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

Example

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

from neomaril_codex.base import NeomarilExecution
from neomaril_codex.model import NeomarilModelClient

def get_execution_status(password, data_path):
    client = BaseNeomarilClient(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') → dict[source]#

Gets the output of the execution.

Parameters:

  • path (str) – Path of the result file. Default value is ‘./’

  • filename (str) – Name of the result file. Default value is ‘output.zip’

Raises:

ExecutionError – Execution unavailable

Returns:

Returns the path for the result file.

Return type:

dict

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

Waits the execution until is no longer running

Example

>>> model.wait_ready()