TaskApi¶

class TaskApi[source]¶

Bases: supervisely.api.module_api.ModuleApiBase, supervisely.api.module_api.ModuleWithStatus

API for working with Tasks. TaskApi object is immutable.

Parameters

api : Api: API connection to the server.

Usage example

import os
from dotenv import load_dotenv

import supervisely as sly

# Load secrets and create API object from .env file (recommended)
# Learn more here: https://developer.supervisely.com/getting-started/basics-of-authentication
if sly.is_development():
    load_dotenv(os.path.expanduser("~/supervisely.env"))
api = sly.Api.from_env()

# Pass values into the API constructor (optional, not recommended)
# api = sly.Api(server_address="https://app.supervise.ly", token="4r47N...xaTatb")

task_id = 121230
task_info = api.task.get_info_by_id(task_id)

Methods

`create_task_detached`
`delete_unused_checkpoints`	rtype `Dict`
`deploy_model`	rtype `int`
`deploy_model_async`	rtype `int`
`download_import_file`
`get_context`	Get context information by task ID.
`get_field`
`get_fields`
`get_import_files_list`	rtype `Optional`[`Dict`]
`get_info_by_id`	Get Task information by ID.
`get_list`	List of Tasks in the given Workspace.
`get_list_all_pages`	Get list of all or limited quantity entities from the Supervisely server.
`get_list_all_pages_generator`	This generator function retrieves a list of all or a limited quantity of entities from the Supervisely server, yielding batches of entities as they are retrieved
`get_status`	Check status of Task by ID.
`get_training_metrics`
`info_sequence`	Get list of all class field names.
`info_tuple_name`	Get string name of NamedTuple.
`list_checkpoints`
`raise_for_status`	Raise error if Task status is ERROR.
`run_dtl`
`run_inference`
`run_train`
`send_request`
`set_field`	rtype `Dict`
`set_fields`	rtype `Dict`
`set_fields_from_dict`	rtype `Dict`
`set_output_archive`	rtype `Dict`
`set_output_directory`
`set_output_error`	Set custom error message to the task output.
`set_output_file_download`	rtype `Dict`
`set_output_project`	rtype `Dict`
`set_output_report`	rtype `Dict`
`start`	Starts the application task on the agent.
`stop`
`submit_logs`	rtype `None`
`update_meta`	Update given task metadata :type id: `int` :param id: int — task id :type data: `dict` :param data: dict — meta data to update
`upload_dtl_archive`
`upload_files`	rtype `None`
`wait`	Awaiting achievement by given Task of a given status.

Attributes

`MAX_WAIT_ATTEMPTS`	Maximum number of attempts that will be made to wait for a certain condition to be met.
`WAIT_ATTEMPT_TIMEOUT_SEC`	Number of seconds for intervals between attempts.

class PluginTaskType[source]¶

Bases: supervisely.collection.str_enum.StrEnum

PluginTaskType

CUSTOM = 'custom'¶

INFERENCE = 'inference'¶

INFERENCE_RPC = 'inference_rpc'¶

SMART_TOOL = 'smarttool'¶

TRAIN = 'train'¶

class RestartPolicy[source]¶

Bases: supervisely.collection.str_enum.StrEnum

RestartPolicy

NEVER = 'never'¶

ON_ERROR = 'on_error'¶

class Status[source]¶

Bases: supervisely.collection.str_enum.StrEnum

Status

CONSUMED = 'consumed'¶: Application is consumed by an agent

DEPLOYED = 'deployed'¶: Only for Plugins

ERROR = 'error'¶: Application has finished with an error

FINISHED = 'finished'¶: Application has finished successfully

QUEUED = 'queued'¶: Application is queued for execution

STARTED = 'started'¶: Application has been started

STOPPED = 'stopped'¶: Application has been stopped

TERMINATING = 'terminating'¶: Application is being terminated

create_task_detached(workspace_id, task_type=None)[source]¶

delete_unused_checkpoints(task_id)[source]¶

Return type: Dict

deploy_model(agent_id, model_id)[source]¶

Return type: int

deploy_model_async(agent_id, model_id)[source]¶

Return type: int

download_import_file(id, file_path, save_path)[source]¶

get_context(id)[source]¶

Get context information by task ID.

Parameters

id : int: Task ID in Supervisely.

Returns

Context information in dict format

Return type

dict

Usage example

import supervisely as sly

task_id = 121230

os.environ['SERVER_ADDRESS'] = 'https://app.supervise.ly'
os.environ['API_TOKEN'] = 'Your Supervisely API Token'
api = sly.Api.from_env()

context = api.task.get_context(task_id)
print(context)
# Output: {
#     "team": {
#         "id": 16087,
#         "name": "alexxx"
#     },
#     "workspace": {
#         "id": 23821,
#         "name": "my_super_workspace"
#     }
# }

get_field(task_id, field)[source]¶

get_fields(task_id, fields)[source]¶

get_import_files_list(id)[source]¶

Return type: Optional[Dict]

get_info_by_id(id)[source]¶

Get Task information by ID.

Parameters

id : int: Task ID in Supervisely.

Returns

Information about Task.

Return type

NamedTuple

Usage example

import supervisely as sly

task_id = 121230

os.environ['SERVER_ADDRESS'] = 'https://app.supervise.ly'
os.environ['API_TOKEN'] = 'Your Supervisely API Token'
api = sly.Api.from_env()

task_info = api.task.get_info_by_id(task_id)
print(task_info)
# Output: {
#     "id": 121230,
#     "workspaceId": 23821,
#     "description": "",
#     "type": "clone",
#     "status": "finished",
#     "startedAt": "2019-12-19T12:13:09.702Z",
#     "finishedAt": "2019-12-19T12:13:09.701Z",
#     "userId": 16154,
#     "meta": {
#         "input": {
#             "model": {
#                 "id": 1849
#             },
#             "isExternal": true,
#             "pluginVersionId": 84479
#         },
#         "output": {
#             "model": {
#                 "id": 12380
#             },
#             "pluginVersionId": 84479
#         }
#     },
#     "settings": {},
#     "agentName": null,
#     "userLogin": "alexxx",
#     "teamId": 16087,
#     "agentId": null
# }

get_list(workspace_id, filters=None)[source]¶

List of Tasks in the given Workspace.

Parameters

workspace_id : int: Workspace ID.
filters : List[dict], optional: List of params to sort output Projects.

Returns

List of Tasks with information for the given Workspace.

Return type

List[NamedTuple]

Usage example

import supervisely as sly

workspace_id = 23821

os.environ['SERVER_ADDRESS'] = 'https://app.supervise.ly'
os.environ['API_TOKEN'] = 'Your Supervisely API Token'
api = sly.Api.from_env()

task_infos = api.task.get_list(workspace_id)

task_infos_filter = api.task.get_list(23821, filters=[{'field': 'id', 'operator': '=', 'value': 121230}])
print(task_infos_filter)
# Output: [
#     {
#         "id": 121230,
#         "type": "clone",
#         "status": "finished",
#         "startedAt": "2019-12-19T12:13:09.702Z",
#         "finishedAt": "2019-12-19T12:13:09.701Z",
#         "meta": {
#             "input": {
#                 "model": {
#                     "id": 1849
#                 },
#                 "isExternal": true,
#                 "pluginVersionId": 84479
#             },
#             "output": {
#                 "model": {
#                     "id": 12380
#                 },
#                 "pluginVersionId": 84479
#             }
#         },
#         "description": ""
#     }
# ]

get_list_all_pages(method, data, progress_cb=None, convert_json_info_cb=None, limit=None, return_first_response=False)¶

Get list of all or limited quantity entities from the Supervisely server.

Parameters

method : str: Request method name
data : dict: Dictionary with request body info
progress_cb : Progress, optional: Function for tracking download progress.
convert_json_info_cb : Callable, optional: Function for convert json info
limit : int, optional: Number of entity to retrieve
return_first_response : bool, optional: Specify if return first response

get_list_all_pages_generator(method, data, progress_cb=None, convert_json_info_cb=None, limit=None, return_first_response=False)¶

This generator function retrieves a list of all or a limited quantity of entities from the Supervisely server, yielding batches of entities as they are retrieved

Parameters

method : str: Request method name
data : dict: Dictionary with request body info
progress_cb : Progress, optional: Function for tracking download progress.
convert_json_info_cb : Callable, optional: Function for convert json info
limit : int, optional: Number of entity to retrieve
return_first_response : bool, optional: Specify if return first response

get_status(task_id)[source]¶

Check status of Task by ID.

Parameters

id : int: Task ID in Supervisely.

Returns

Status object

Return type

Status

Usage example

import supervisely as sly

task_id = 121230

os.environ['SERVER_ADDRESS'] = 'https://app.supervise.ly'
os.environ['API_TOKEN'] = 'Your Supervisely API Token'
api = sly.Api.from_env()

task_status = api.task.get_status(task_id)
print(task_status)
# Output: finished

get_training_metrics(task_id)[source]¶

static info_sequence()¶: Get list of all class field names.

static info_tuple_name()¶: Get string name of NamedTuple.

list_checkpoints(task_id)[source]¶

raise_for_status(status)[source]¶

Raise error if Task status is ERROR.

Parameters

status : Status: Status object.

Returns

None

Return type

NoneType

run_dtl(workspace_id, dtl_graph, agent_id=None)[source]¶

run_inference(agent_id, input_project_id, input_model_id, result_project_name, inference_config=None)[source]¶

run_train(agent_id, input_project_id, input_model_id, result_nn_name, train_config=None)[source]¶

send_request(task_id, method, data, context={}, skip_response=False, timeout=60, outside_request=True, retries=10, raise_error=False)[source]¶

set_field(task_id, field, payload, append=False, recursive=False)[source]¶

Return type: Dict

set_fields(task_id, fields)[source]¶

Return type: Dict

set_fields_from_dict(task_id, d)[source]¶

Return type: Dict

set_output_archive(task_id, file_id, file_name, file_url=None)[source]¶

Return type: Dict

set_output_directory(task_id, file_id, directory_path)[source]¶

set_output_error(task_id, title, description=None, show_logs=True)[source]¶

Set custom error message to the task output.

Parameters

task_id : int: Application task ID.
title : str: Error message to be displayed in the task output.
description : Optional[str]: Description to be displayed in the task output.
show_logs : Optional[bool], default True: If True, the link to the task logs will be displayed in the task output.

Returns

Response JSON.

Return type

Dict

Usage example

import os
from dotenv import load_dotenv

import supervisely as sly

# Load secrets and create API object from .env file (recommended)
# Learn more here: https://developer.supervisely.com/getting-started/basics-of-authentication
if sly.is_development():
   load_dotenv(os.path.expanduser("~/supervisely.env"))
api = sly.Api.from_env()

task_id = 12345
title = "Something went wrong"
description = "Please check the task logs"
show_logs = True
api.task.set_output_error(task_id, title, description, show_logs)

set_output_file_download(task_id, file_id, file_name, file_url=None, download=True)[source]¶

Return type: Dict

set_output_project(task_id, project_id, project_name=None)[source]¶

Return type: Dict

set_output_report(task_id, file_id, file_name)[source]¶

Return type: Dict

start(agent_id, app_id=None, workspace_id=None, description='application description', params=None, log_level='info', users_ids=None, app_version='', is_branch=False, task_name='pythonSpawned', restart_policy='never', proxy_keep_url=False, module_id=None, redirect_requests={})[source]¶

Starts the application task on the agent.

Parameters

agent_id : int: Agent ID. Can be obtained from TeamCluster page in UI.
app_id : int, optional: Deprecated. Use module_id instead.
workspace_id : int, optional: Workspace ID where the task will be created.
description : str, optional: Task description which will be shown in UI.
params : Dict[str, Any], optional: Task parameters which will be passed to the application, check the code example below for more details.
log_level : Literal["info", "debug", "warning", "error"], optional: Log level for the application.
users_ids : List[int], optional: List of user IDs for which will be created an instance of the application. For each user a separate task will be created.
app_version : str, optional: Application version e.g. “v1.0.0” or branch name e.g. “dev”.
is_branch : bool, optional: If the application version is a branch name, set this parameter to True.
task_name : str, optional: Task name which will be shown in UI.
restart_policy : Literal["never", "on_error"], optional: when the task should be restarted: never or if error occurred.
proxy_keep_url : bool, optional: For internal usage only.
module_id : int, optional: Module ID. Can be obtained from the apps page in UI.
redirect_requests : Dict[str, int], optional: For internal usage only in Develop and Debug mode.

Returns

Task information in JSON format.

Return type

Dict[str, Any]

Usage example

import supervisely as sly

app_slug = "supervisely-ecosystem/export-to-supervisely-format"
module_id = api.app.get_ecosystem_module_id(app_slug)
module_info = api.app.get_ecosystem_module_info(module_id)

project_id = 12345
agent_id = 12345
workspace_id = 12345

params = module_info.get_arguments(images_project=project_id)

session = api.app.start(
    agent_id=agent_id,
    module_id=module_id,
    workspace_id=workspace_id,
    task_name="Prepare download link",
    params=params,
    app_version="dninja",
    is_branch=True,
)

stop(id)[source]¶

submit_logs(logs)[source]¶

Return type: None

update_meta(id, data, agent_storage_folder=None, relative_app_dir=None)[source]¶: Update given task metadata :type id: int :param id: int — task id :type data: dict :param data: dict — meta data to update

upload_dtl_archive(task_id, archive_path, progress_cb=None)[source]¶

upload_files(task_id, abs_paths, names, progress_cb=None)[source]¶

Return type: None

wait(id, target_status, wait_attempts=None, wait_attempt_timeout_sec=None)[source]¶

Awaiting achievement by given Task of a given status.

Parameters

id : int: Task ID in Supervisely.
target_status : Status: Status object(status of task we expect to destinate).
wait_attempts : int, optional: The number of attempts to determine the status of the task that we are waiting for.
wait_attempt_timeout_sec : int, optional: Number of seconds for intervals between attempts(raise error if waiting time exceeded).

Returns

True if the desired status is reached, False otherwise

Return type

bool