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'
DEPLOYED = 'deployed'
ERROR = 'error'
FINISHED = 'finished'
QUEUED = 'queued'
STARTED = 'started'
STOPPED = 'stopped'
TERMINATING = 'terminating'
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)[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