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

convert_info_to_json

_convert_info_to_json

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_list_idx_page_async

Get the list of items for a given page number.

get_list_page_generator_async

Yields list of images in dataset asynchronously page by page.

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_experiment

Sets output for the task with experiment info.

set_output_file_download

rtype

Dict

set_output_project

rtype

Dict

set_output_report

rtype

Dict

set_output_text

Set custom text message to the task output.

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

update_status

Sets the specified status for the task.

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

classmethod convert_info_to_json(info)

_convert_info_to_json

Return type

Dict

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.supervisely.com'
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.supervisely.com'
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": {
#         "app": {
#             "id": 10370,
#             "name": "Auto Import",
#             "version": "test-branch",
#             "isBranch": true,
#         },
#         "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.supervisely.com'
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

async get_list_idx_page_async(method, data)

Get the list of items for a given page number. Page number is specified in the data dictionary.

Parameters
method : str

Method to call for listing items.

data : dict

Data to pass to the API method.

Returns

List of items.

Return type

Tuple[int, List[NamedTuple]]

async get_list_page_generator_async(method, data, pages_count=None, semaphore=None)

Yields list of images in dataset asynchronously page by page.

Parameters
method : str

Method to call for listing items.

data : dict

Data to pass to the API method.

pages_count : int, optional

Preferred number of pages to retrieve if used with a per_page limit. Will be automatically adjusted if the pagesCount differs from the requested number.

semaphore : asyncio.Semaphore, optional

Semaphore for limiting the number of simultaneous requests.

kwargs

Additional arguments.

Returns

List of images in dataset.

Return type

AsyncGenerator[List[ImageInfo]]

Usage example
import supervisely as sly
import asyncio

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

method = 'images.list'
data = {
    'datasetId': 123456
}

loop = sly.utils.get_or_create_event_loop()
images = loop.run_until_complete(api.image.get_list_generator_async(method, data))
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.supervisely.com'
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_experiment(task_id, experiment_info)[source]

Sets output for the task with experiment info.

Parameters
task_id : int

Task ID in Supervisely.

experiment_info : dict

Experiment info from TrainApp.

Returns

None

Return type

NoneType

Example of experiment_info:

experiment_info = {

‘experiment_name’: ‘247_Lemons_RT-DETRv2-M’, ‘framework_name’: ‘RT-DETRv2’, ‘model_name’: ‘RT-DETRv2-M’, ‘task_type’: ‘object detection’, ‘project_id’: 76, ‘task_id’: 247, ‘model_files’: {‘config’: ‘model_config.yml’}, ‘checkpoints’: [‘checkpoints/best.pth’, ‘checkpoints/checkpoint0025.pth’, ‘checkpoints/checkpoint0050.pth’, ‘checkpoints/last.pth’], ‘best_checkpoint’: ‘best.pth’, ‘export’: {‘ONNXRuntime’: ‘export/best.onnx’}, ‘app_state’: ‘app_state.json’, ‘model_meta’: ‘model_meta.json’, ‘train_val_split’: ‘train_val_split.json’, ‘train_size’: 4, ‘val_size’: 2, ‘hyperparameters’: ‘hyperparameters.yaml’, ‘hyperparameters_id’: 45234, ‘artifacts_dir’: ‘/experiments/76_Lemons/247_RT-DETRv2/’, ‘datetime’: ‘2025-01-22 18:13:43’, ‘evaluation_report_id’: 12961, ‘evaluation_report_link’: ‘https://app.supervisely.com/model-benchmark?id=12961’, ‘evaluation_metrics’: {

‘mAP’: 0.994059405940594, ‘AP50’: 1.0, ‘AP75’: 1.0, ‘f1’: 0.9944444444444445, ‘precision’: 0.9944444444444445, ‘recall’: 0.9944444444444445, ‘iou’: 0.9726227736959404, ‘classification_accuracy’: 1.0, ‘calibration_score’: 0.8935745942476048, ‘f1_optimal_conf’: 0.500377893447876, ‘expected_calibration_error’: 0.10642540575239527, ‘maximum_calibration_error’: 0.499622106552124

}, ‘primary_metric’: ‘mAP’ ‘logs’: {

‘type’: ‘tensorboard’, ‘link’: ‘/experiments/76_Lemons/247_RT-DETRv2/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, description='Report')[source]
Return type

Dict

set_output_text(task_id, title, description=None, show_logs=False, zmdi_icon='zmdi-comment-alt-text', icon_color='#33c94c', background_color='#d9f7e4')[source]

Set custom text message to the task output.

Parameters
task_id : int

Application task ID.

title : str

Text message to be displayed in the task output.

description : Optional[str]

Description to be displayed in the task output.

show_logs : Optional[bool], default False

If True, the link to the task logs will be displayed in the task output.

zmdi_icon : Optional[str], default "zmdi-comment-alt-text"

Icon class name from Material Design Icons (ZMDI).

icon_color : Optional[str], default "#33c94c" (nearest Duron Jolly Green)

Icon color in HEX format.

background_color : Optional[str], default "#d9f7e4" (Cosmic Latte)

Background color in HEX format.

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 = "Task is finished"
api.task.set_output_text(task_id, title)
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={}, limit_by_workspace=False)[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.

limit_by_workspace : bool, optional

If set to True tasks will be only visible inside of the workspace with specified workspace_id.

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

update_status(task_id, status)[source]

Sets the specified status for the task.

Parameters
task_id : int

Task ID in Supervisely.

status : One of the values from Status, e.g. Status.FINISHED, Status.ERROR, etc.

Task status to set.

Raises

ValueError – If the status value is not allowed.

Return type

None

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