Task API

class supervisely_lib.api.task_api.TaskApi(api)[source]

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

API for working with Tasks. TaskApi object is immutable.

Parameters

api (Api) – API connection to the server.

Usage example
# You can connect to API directly
address = 'https://app.supervise.ly/'
token = 'Your Supervisely API Token'
api = sly.Api(address, token)

# Or you can use API from environment
os.environ['SERVER_ADDRESS'] = 'https://app.supervise.ly'
os.environ['API_TOKEN'] = 'Your Supervisely API Token'
api = sly.Api.from_env()

task_id = 121230
task_info = api.task.get_info_by_id(task_id)
class RestartPolicy(value)[source]

Bases: supervisely_lib.collection.str_enum.StrEnum

An enumeration.

NEVER = 'never'
ON_ERROR = 'on_error'
class PluginTaskType(value)[source]

Bases: supervisely_lib.collection.str_enum.StrEnum

An enumeration.

TRAIN = 'train'
INFERENCE = 'inference'
INFERENCE_RPC = 'inference_rpc'
SMART_TOOL = 'smarttool'
CUSTOM = 'custom'
class Status(value)[source]

Bases: supervisely_lib.collection.str_enum.StrEnum

An enumeration.

QUEUED = 'queued'
CONSUMED = 'consumed'
STARTED = 'started'
DEPLOYED = 'deployed'
ERROR = 'error'
FINISHED = 'finished'
TERMINATING = 'terminating'
STOPPED = 'stopped'
get_list(workspace_id: int, filters: Optional[List[dict]] = None) → List[NamedTuple][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
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_info_by_id(id: int) → NamedTuple[source]

Get Task information by ID.

Parameters

id (int) – Task ID in Supervisely.

Returns

Information about Task.

Return type

NamedTuple

Usage example
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_status(task_id: int)supervisely_lib.api.task_api.TaskApi.Status[source]

Check status of Task by ID.

Parameters

id (int) – Task ID in Supervisely.

Returns

Status object

Return type

Status

Usage example
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
raise_for_status(status: supervisely_lib.api.task_api.TaskApi.Status) → None[source]

Raise error if Task status is ERROR.

Parameters

status (Status) – Status object.

Returns

None

Return type

NoneType

wait(id: int, target_status: supervisely_lib.api.task_api.TaskApi.Status, wait_attempts: Optional[int] = None, wait_attempt_timeout_sec: Optional[int] = None) → bool[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

upload_dtl_archive(task_id, archive_path, progress_cb=None)[source]
get_context(id: int) → dict[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
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"
#     }
# }
run_dtl(workspace_id, dtl_graph, agent_id=None)[source]
run_train(agent_id, input_project_id, input_model_id, result_nn_name, train_config=None)[source]
run_inference(agent_id, input_project_id, input_model_id, result_project_name, inference_config=None)[source]
get_training_metrics(task_id)[source]
deploy_model(agent_id, model_id)[source]
deploy_model_async(agent_id, model_id)[source]
stop(id)[source]
get_import_files_list(id)[source]
download_import_file(id, file_path, save_path)[source]
create_task_detached(workspace_id, task_type: Optional[str] = None)[source]
submit_logs(logs)[source]
upload_files(task_id, abs_paths, names, progress_cb=None)[source]
set_fields(task_id, fields)[source]
set_field(task_id, field, payload, append=False, recursive=False)[source]
get_fields(task_id, fields: list) → dict[source]
get_field(task_id, field)[source]
list_checkpoints(task_id)[source]
delete_unused_checkpoints(task_id)[source]
set_output_project(task_id, project_id, project_name=None)[source]
set_output_report(task_id, file_id, file_name)[source]
set_output_archive(task_id, file_id, file_name, file_url=None)[source]
send_request(task_id, method, data, context={}, skip_response=False)[source]
set_output_directory(task_id, file_id, directory_path)[source]