Api¶

class Api(server_address=None, token=None, retry_count=10, retry_sleep_sec=None, external_logger=None, ignore_task_id=False, api_server_address=None, check_instance_version=False)[source]¶

Bases: object

Main Supervisely API client (aggregates sub-APIs and handles auth, retries and common configuration).

Parameters:

server_address : str¶: Address of the server.
token : str¶: Unique secret token associated with your agent.
retry_count : int, optional¶: The number of attempts to connect to the server.
retry_sleep_sec : int, optional¶: The number of seconds to delay between attempts to connect to the server.
external_logger : logger, optional¶: Logger class object.
ignore_task_id : bool, optional¶
api_server_address : str, optional¶: Address of the API server.
check_instance_version : bool or str, optional¶: Check if the given version is lower or equal to the current Supervisely instance version. If set to True, will try to read the version from the environment variable “MINIMUM_INSTANCE_VERSION_FOR_SDK”. If set to a string, will use this string as the version to check. If set to False, will skip the check.

Raises:

ValueError – if token is None or it length != 128

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

# Or you can pass values into the API constructor (optional, not recommended)
server_address = "https://app.supervisely.com"
token = "4r47N...xaTatb"
api = sly.Api(server_address, token)

Methods

`add_additional_field`	Add given key and value to additional_fields dictionary.
`add_header`	Add given key and value to headers dictionary.
`from_credentials`	Create Api object using credentials and optionally save them to ".env" file with overriding environment variables.
`from_env`	Initialize API use environment variables.
`get`	Performs GET request to server with given parameters.
`get_default_semaphore`	Get default global API semaphore for async requests.
`get_httpx`	Performs GET request to server with given parameters.
`is_version_supported`	Check if the given version is lower or equal to the current Supervisely instance version.
`normalize_server_address`
`parse_error`	Processes error from response.
`pop_header`
`post`	Performs POST request to server with given parameters.
`post_async`	Performs POST request to server with given parameters using httpx.
`post_httpx`	Performs POST request to server with given parameters using httpx.
`set_semaphore_size`	Set the global API semaphore with the given size.
`stream`	Performs streaming GET or POST request to server with given parameters.
`stream_async`	Performs asynchronous streaming GET or POST request to server with given parameters.

Attributes

`api_server_address`	Get API server address.
`instance_version`	Return Supervisely instance version, e.g. "6.9.13".
`semaphore`	Get the global API semaphore for async requests.

classmethod from_credentials(server_address, login, password, override=False, env_file='/home/docs/supervisely.env', check_instance_version=False)[source]¶

Create Api object using credentials and optionally save them to “.env” file with overriding environment variables. If “.env” file already exists, backup will be created automatically. All backups will be stored in the same directory with postfix “_YYYYMMDDHHMMSS”. You can have not more than 5 last backups. This method can be used also to update “.env” file.

Parameters:

server_address : str¶: Supervisely server url.
login : str¶: User login.
password : str¶: User password.
override : bool, optional¶: If False, return Api object. If True, additionally create “.env” file or overwrite existing (backup file will be created automatically), and override environment variables.
env_file : str, optional¶: Path to your .env file.
check_instance_version : bool or str, optional¶: Check if the given version is lower or equal to the current version of the Supervisely instance.

Returns:

Api object

Return type:

Api

Usage Example:

import supervisely as sly

server_address = 'https://app.supervisely.com'
login = 'user'
password = 'pass'

api = sly.Api.from_credentials(server_address, login, password)

classmethod from_env(retry_count=10, ignore_task_id=False, env_file='/home/docs/supervisely.env', check_instance_version=False)[source]¶

Initialize API use environment variables.

Parameters:

retry_count : int¶: The number of attempts to connect to the server.
ignore_task_id : bool¶
env_file : str¶: Path to your .env file.
check_instance_version : bool or str, optional¶: Check if the given version is lower or equal to the current version of the Supervisely instance.

Returns:

Api object

Return type:

Api

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

static parse_error(response, default_error='Error', default_message='please, contact administrator')[source]¶

Processes error from response.

Parameters:

response¶: Request object.
default_error='Error'¶: Error description.
default_message='please, contact administrator'¶: Message to user.

Returns:

Number of error and message about curren connection mistake

Return type:

int, str

add_additional_field(key, value)[source]¶

Add given key and value to additional_fields dictionary.

Parameters:

key : str¶: New key.
value : str¶: New value.

Returns:

None

Return type:

None

add_header(key, value)[source]¶

Add given key and value to headers dictionary.

Parameters:

key : str¶: New key.
value : str¶: New value.

Raises:

RuntimeError – if key is already set

Returns:

None

Return type:

None

get(method, params, retries=None, stream=False, use_public_api=True, data=None)[source]¶

Performs GET request to server with given parameters.

Parameters:

method : str¶
params : dict¶: Dictionary to send in the body of the Request.
retries : int, optional¶: The number of attempts to connect to the server.
stream : bool, optional¶: Define, if you’d like to get the raw socket response from the server.
use_public_api : bool, optional¶
data : dict, optional¶: Dictionary to send in the body of the Request.

Returns:

Response object

Return type:

Response

get_default_semaphore()[source]¶

Get default global API semaphore for async requests. If the semaphore is not set, it will be initialized.

During initialization, the semaphore size will be set from the environment variable SUPERVISELY_ASYNC_SEMAPHORE. If the environment variable is not set, the default value will be set based on the server address. Depending on the server address, the semaphore size will be set to 10 for HTTPS and 5 for HTTP.

Returns:: Semaphore object.
Return type:: asyncio.Semaphore

get_httpx(method, params, retries=None, use_public_api=True, timeout=60)[source]¶

Performs GET request to server with given parameters.

Parameters:

method : str¶: Method name.
params : httpx._types.QueryParamTypes¶: URL query parameters.
retries : int, optional¶: The number of attempts to connect to the server.
use_public_api : bool, optional¶: Define if public API should be used. Default is True.
timeout : float, optional¶: Overall timeout for the request.

Returns:

Response object

Return type:

Response

is_version_supported(version=None)[source]¶

Check if the given version is lower or equal to the current Supervisely instance version. If the version omitted, will try to read it from the environment variable “MINIMUM_INSTANCE_VERSION_FOR_SDK”. If the version is lower or equal, return True, otherwise False. If the version of the instance cannot be determined, return False.

Parameters:

version : Optional[str], e.g. "6.9.13"¶: Version to check.

Returns:

True if the given version is lower or equal to the current Supervisely instance version, otherwise False.

Return type:

bool

Usage Example:

import supervisely as sly

api = sly.Api(server_address='https://app.supervisely.com', token='4r47N...xaTatb')
version_to_check = "6.9.13"
print(api.is_version_supported(version_to_check))
# Output: True

post(method, data, retries=None, stream=False, raise_error=False)[source]¶

Performs POST request to server with given parameters.

Parameters:

method : str¶: Method name.
data : dict¶: Dictionary to send in the body of the Request.
retries : int, optional¶: The number of attempts to connect to the server.
stream : bool, optional¶: Define, if you’d like to get the raw socket response from the server.
raise_error : bool, optional¶: Define, if you’d like to raise error if connection is failed. Retries will be ignored.

Returns:

Response object

Return type:

Response

async post_async(method, json=None, content=None, files=None, params=None, headers=None, retries=None, raise_error=False, timeout=60)[source]¶

Performs POST request to server with given parameters using httpx.

Parameters:

method : str¶: Method name.
json : dict, optional¶: Dictionary to send in the body of request.
content : bytes or dict, optional¶: Bytes with data content or dictionary with params.
files : dict, optional¶: Files to send in the body of request.
params : str, bytes, optional¶: URL query parameters.
headers : dict, optional¶: Custom headers to include in the request.
retries : int, optional¶: The number of attempts to connect to the server.
raise_error : bool, optional¶: Define, if you’d like to raise error if connection is failed.
timeout : float, optional¶: Overall timeout for the request.

Returns:

Response object

Return type:

httpx.Response

post_httpx(method, json=None, content=None, files=None, params=None, headers=None, retries=None, raise_error=False, timeout=60)[source]¶

Performs POST request to server with given parameters using httpx.

Parameters:

method : str¶: Method name.
json : dict, optional¶: Dictionary to send in the body of request.
content : bytes or dict, optional¶: Bytes with data content or dictionary with params.
files : dict, optional¶: Files to send in the body of request.
params : str, bytes, optional¶: URL query parameters.
headers : dict, optional¶: Custom headers to include in the request.
retries : int, optional¶: The number of attempts to connect to the server.
raise_error : bool, optional¶: Define, if you’d like to raise error if connection is failed.
timeout : float, optional¶: Overall timeout for the request.

Returns:

Response object

Return type:

httpx.Response

set_semaphore_size(size=None)[source]¶

Set the global API semaphore with the given size. Will replace the existing semaphore. If the size is not set, will set from the environment variable SUPERVISELY_ASYNC_SEMAPHORE. If the environment variable is not set, will set the default value.

Parameters:

size : int, optional¶: Size of the semaphore.

stream(method, method_type, data, headers=None, retries=None, range_start=None, range_end=None, raise_error=False, chunk_size=8192, use_public_api=True, timeout=60)[source]¶

Performs streaming GET or POST request to server with given parameters. Multipart is not supported.

Parameters:

method : str¶: Method name for the request.
method_type : str¶: Request type (‘GET’ or ‘POST’).
data : bytes or dict¶: Bytes with data content or dictionary with params.
headers : dict, optional¶: Custom headers to include in the request.
retries : int, optional¶: The number of retry attempts.
range_start : int, optional¶: Start byte position for streaming.
range_end : int, optional¶: End byte position for streaming.
raise_error : bool, optional¶: If True, raise raw error if the request fails.
chunk_size : int, optional¶: Size of the chunks to stream.
use_public_api : bool, optional¶: Define if public API should be used.
timeout : float, optional¶: Overall timeout for the request.

Returns:

Generator object.

Return type:

Generator

async stream_async(method, method_type, data, headers=None, retries=None, range_start=None, range_end=None, chunk_size=8192, use_public_api=True, timeout=60, **kwargs)[source]¶

Performs asynchronous streaming GET or POST request to server with given parameters. Yield chunks of data and hash of the whole content to check integrity of the data stream.

Parameters:

method : str¶: Method name for the request.
method_type : str¶: Request type (‘GET’ or ‘POST’).
data : bytes or dict¶: Bytes with data content or dictionary with params.
headers : dict, optional¶: Custom headers to include in the request.
retries : int, optional¶: The number of retry attempts.
range_start : int, optional¶: Start byte position for streaming.
range_end : int, optional¶: End byte position for streaming.
chunk_size : int, optional¶: Size of the chunk to read from the stream. Default is 8192.
use_public_api : bool, optional¶: Define if public API should be used.
timeout : float, optional¶: Overall timeout for the request.

Returns:

Async generator object.

Return type:

AsyncGenerator

property api_server_address : str¶

Get API server address.

Returns:

API server address.

Return type:

str

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()
print(api.api_server_address)
# Output: 'https://app.supervisely.com/public/api'

property instance_version : str¶

Return Supervisely instance version, e.g. “6.9.13”. If the version cannot be determined, return “unknown”.

Returns:

Supervisely instance version or “unknown” if the version cannot be determined.

Return type:

str

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()
print(api.instance_version)
# Output:
# '6.9.13'

property semaphore : asyncio.locks.Semaphore¶

Get the global API semaphore for async requests.

Returns:: Semaphore object.
Return type:: asyncio.Semaphore