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
- rtype
- rtype
- rtype
Get context information by task ID.
Get Task information by ID.
List of Tasks in the given Workspace.
Get list of all or limited quantity entities from the Supervisely server.
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 the list of items for a given page number.
Yields list of images in dataset asynchronously page by page.
Check status of Task by ID.
Get list of all class field names.
Get string name of NamedTuple.
Raise error if Task status is ERROR.
- rtype
- rtype
- rtype
- rtype
Set custom error message to the task output.
Sets output for the task with experiment info.
- rtype
- rtype
- rtype
Set custom text message to the task output.
Starts the application task on the agent.
- rtype
Update given task metadata :type id:
int
:param id: int — task id :type data:dict
:param data: dict — meta data to updateSets the specified status for the task.
- rtype
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'
¶
-
CUSTOM =
- class RestartPolicy[source]¶
Bases:
supervisely.collection.str_enum.StrEnum
RestartPolicy
-
NEVER =
'never'
¶
-
ON_ERROR =
'on_error'
¶
-
NEVER =
- 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
-
CONSUMED =
- 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
- 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_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
- 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.
-
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 thepagesCount
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
- 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
- static info_sequence()¶
Get list of all class field names.
- static info_tuple_name()¶
Get string name of NamedTuple.
- raise_for_status(status)[source]¶
Raise error if Task status is ERROR.
- Parameters
- status : Status
Status object.
- Returns
None
- Return type
NoneType
-
run_inference(agent_id, input_project_id, input_model_id, result_project_name, inference_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_output_error(task_id, title, description=
None
, show_logs=True
)[source]¶ Set custom error message to the task output.
- Parameters
- 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
- 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
-
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, )
-
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
- Raises
ValueError – If the status value is not allowed.
- Return type
-
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