TaskApi¶

Bases: ModuleApiBase, ModuleWithStatus

API for working with tasks.

Parameters:

api¶: Api object to use for API connection.

Usage Example:

import supervisely as sly
api = sly.Api.from_env()
task_info = api.task.get_info_by_id(task_id)

Methods

`convert_info_to_json`	Convert information about an entity to a dictionary.
`get_context`	Get context information by Application Task ID.
`get_field`	get_field
`get_fields`	get_fields
`get_info_by_id`	Get Application Task information by ID.
`get_list`	List of Application 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_logs`	Returns the log entries for the given task ID in JSON format.
`get_status`	Check status of Application Task by ID.
`info_sequence`	Get list of all class field names.
`info_tuple_name`	Get string name of NamedTuple.
`is_ready`	Check if the task is ready.
`is_running`	Check if the task is running.
`raise_for_status`	Raise error if Application Task status is ERROR.
`send_request`	send_request
`set_field`	set_field
`set_fields`	set_fields
`set_fields_from_dict`	set_fields_from_dict
`set_output_archive`	set_output_archive
`set_output_directory`	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`	set_output_file_download
`set_output_project`	set_output_project
`set_output_report`	set_output_report
`set_output_text`	Set custom text message to the task output.
`start`	Starts the Application Task on the agent.
`stop`	stop
`submit_logs`	submit_logs
`update_meta`	Update given task metadata :param id: int - task id :param data: dict - meta data to update
`update_status`	Sets the specified status for the task.
`upload_files`	upload_files
`wait`	Awaiting achievement by given Application 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(*values)[source]¶

Bases: StrEnum

Task type for plugin-based tasks (train/inference/smarttool/etc.).

classmethod has_value(value)¶

Check if a value is in the enum.

Parameters:

value : str¶: Value to check.

Returns:

True if value is in the enum, False otherwise.

Return type:

bool

classmethod values()¶: Get all values of the enum.

CUSTOM = 'custom'¶

INFERENCE = 'inference'¶

INFERENCE_RPC = 'inference_rpc'¶

SMART_TOOL = 'smarttool'¶

TRAIN = 'train'¶

class RestartPolicy(*values)[source]¶

Bases: StrEnum

Task restart policy used for app deployments.

classmethod has_value(value)¶

Check if a value is in the enum.

Parameters:

value : str¶: Value to check.

Returns:

True if value is in the enum, False otherwise.

Return type:

bool

classmethod values()¶: Get all values of the enum.

NEVER = 'never'¶

ON_ERROR = 'on_error'¶

class Status(*values)[source]¶

Bases: StrEnum

Task lifecycle status values returned by the API.

classmethod has_value(value)¶

Check if a value is in the enum.

Parameters:

value : str¶: Value to check.

Returns:

True if value is in the enum, False otherwise.

Return type:

bool

classmethod values()¶: Get all values of the enum.

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 information about an entity to a dictionary.

Parameters:

info : NamedTuple¶: Information about the entity.

Returns:

Dictionary with information about the entity.

Return type:

dict

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

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

get_context(id)[source]¶

Get context information by Application Task ID.

Parameters:

id : int¶: Task ID in Supervisely.

Returns:

Context information in dict format

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 = 121230
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_info_by_id(id)[source]¶

Get Application Task information by ID.

Parameters:

id : int¶: Task ID in Supervisely.

Returns:

Information about Task.

Return type:

NamedTuple

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 = 121230
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 Application 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 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()

workspace_id = 23821
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

Returns:

List of entities.

Return type:

List[dict]

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=None¶: Semaphore for limiting the number of simultaneous requests.

Returns:

List of images in dataset.

Return type:

AsyncGenerator[List[ImageInfo]]

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

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_logs(task_id, since_time=None, limit=None)[source]¶

Returns the log entries for the given task ID in JSON format.

Parameters:

task_id : int¶: Task ID in Supervisely.
since_time : Optional[str]¶: Optional ISO 8601 formatted timestamp to filter logs that were created after the specified time.
limit : Optional[int]¶: Optional integer to limit the number of log entries returned.

Returns:

List of log entries in JSON format.

Return type:

List[Dict]

Usage Example:

since_time = "2026-03-30T14:28:58.816Z"
limit = 100
task_id = 12345
logs = api.task.get_logs(task_id, since_time=since_time, limit=limit)
for log_entry in logs:
    print(log_entry) # {'level': 'info', 'message': "Added object class... 'timestamp': '2026-03-30T14:28:56.525Z', 'task_id': 58595}

get_status(task_id)[source]¶

Check status of Application Task by ID.

Parameters:

task_id : int¶: Task ID in Supervisely.

Returns:

Status object

Return type:

Status

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 = 121230
task_status = api.task.get_status(task_id)
print(task_status)
# Output: finished

is_ready(task_id)[source]¶

Check if the task is ready.

Parameters:

task_id : int¶: Task ID in Supervisely.

Returns:

True if the task is ready, False otherwise.

Return type:

bool

is_running(task_id)[source]¶

Check if the task is running.

Parameters:

task_id : int¶: Task ID in Supervisely.

Returns:

True if the task is running, False otherwise.

Return type:

bool

raise_for_status(status)[source]¶

Raise error if Application Task status is ERROR.

Parameters:

status : Status¶: Status object.

Returns:

None

Return type:

None

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]¶

set_fields(task_id, fields)[source]¶

set_fields_from_dict(task_id, d)[source]¶

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

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. See class ExperimentInfo.

Returns:

Server response JSON.

Return type:

dict

Usage Example:

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,
    "project_version": {"id": 222, "version": 4},
    "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,
    "train_collection_id": 530,
    "val_collection_id": 531,
    "hyperparameters": "hyperparameters.yaml",
    "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.994, "AP50": 1.0, "AP75": 1.0},  # and other metrics
    "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]¶

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

set_output_report(task_id, file_id, file_name, description='Report')[source]¶

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 set (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, kubernetes_settings=None, multi_user_session=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.
kubernetes_settings=None¶: Kubernetes settings for the application.
multi_user_session : bool, default is False¶: If True, the application session will be created as multi-user. In this case, multiple users will be able to connect to the same application session. All users will have separate application states. Available only for applications that support multi-user sessions.

Returns:

Task information in JSON format.

Return type:

Dict[str, Any]

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

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]¶

update_meta(id, data, agent_storage_folder=None, relative_app_dir=None)[source]¶: Update given task metadata :param id: int - task id :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¶: Task status to set.

Raises:

ValueError – If the status value is not allowed.

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

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

Awaiting achievement by given Application 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