Low-level API

Overview

The low-level API is useful if you need to work directly with REST API, but want to have data validation and syntax assistance from your code editor. The code on this layer is autogenerated.

Code of this component is located in cvat_sdk.api_client.

Example

Let’s see how a task with local files can be created. We will use the basic auth to make things simpler.

from time import sleep
from cvat_sdk.api_client import Configuration, ApiClient, models, apis, exceptions

configuration = Configuration(
    host="http://localhost",
    username='YOUR_USERNAME',
    password='YOUR_PASSWORD',
)

# Enter a context with an instance of the API client
with ApiClient(configuration) as api_client:
    # Parameters can be passed as a plain dict with JSON-serialized data
    # or as model objects (from cvat_sdk.api_client.models), including
    # mixed variants.
    #
    # In case of dicts, keys must be the same as members of models.I<ModelName>
    # interfaces and values must be convertible to the corresponding member
    # value types (e.g. a date or string enum value can be parsed from a string).
    #
    # In case of model objects, data must be of the corresponding
    # models.<ModelName> types.
    #
    # Let's use a dict here. It should look like models.ITaskWriteRequest
    task_spec = {
        'name': 'example task',
        "labels": [{
            "name": "car",
            "color": "#ff00ff",
            "attributes": [
                {
                    "name": "a",
                    "mutable": True,
                    "input_type": "number",
                    "default_value": "5",
                    "values": ["4", "5", "6"]
                }
            ]
        }],
    }

    try:
        # Apis can be accessed as ApiClient class members
        # We use different models for input and output data. For input data,
        # models are typically called like "*Request". Output data models have
        # no suffix.
        (task, response) = api_client.tasks_api.create(task_spec)
    except exceptions.ApiException as e:
        # We can catch the basic exception type, or a derived type
        print("Exception when trying to create a task: %s\n" % e)

    # Here we will use models instead of a dict
    task_data = models.DataRequest(
        image_quality=75,
        client_files=[
            open('image1.jpg', 'rb'),
            open('image2.jpg', 'rb'),
        ],
    )

    # If we pass binary file objects, we need to specify content type.
    # For this endpoint, we don't have response data
    (_, response) = api_client.tasks_api.create_data(task.id,
        data_request=task_data,
        _content_type="multipart/form-data",

        # we can choose to check the response status manually
        # and disable the response data parsing
        _check_status=False, _parse_response=False
    )
    assert response.status == 202, response.msg

    # Wait till task data is processed
    for _ in range(100):
        (status, _) = api_client.tasks_api.retrieve_status(task.id)
        if status.state.value in ['Finished', 'Failed']:
            break
        sleep(0.1)
    assert status.state.value == 'Finished', status.message

    # Update the task object and check the task size
    (task, _) = api_client.tasks_api.retrieve(task.id)
    assert task.size == 4

ApiClient and configuration

The starting point in the low-level API is the cvat_sdk.api_client.ApiClient class. It encapsulates session and connection logic, manages headers and cookies, and provides access to various APIs.

To create an instance of ApiClient, you need to set up a cvat_sdk.api_client.Configuration object and pass it to the ApiClient class constructor. Additional connection-specific options, such as extra headers and cookies can be specified in the class constructor. ApiClient implements the context manager protocol. Typically, you create ApiClient this way:

from cvat_sdk.api_client import ApiClient, Configuration

configuration = Configuration(host="http://localhost")
with ApiClient(configuration) as api_client:
    ...

After creating an ApiClient instance, you can send requests to various server endpoints via *_api member properties and directly, using the rest_client member. Read more about API wrappers below.

Typically, the first thing you do with ApiClient is log in. Read more about authentication options below.

Authentication

CVAT supports 2 authentication options:

• basic auth, with your username and password
• token auth, with your API key

Token auth requires a token, which can be obtained after performing the basic auth.

The low-level API supports 2 ways of authentication. You can specify authentication parameters in the Configuration object:

configuration = Configuration(
    username='YOUR_USERNAME',
    password='YOUR_PASSWORD',
)

configuration = Configuration(
    api_key={
        "sessionAuth": "<sessionid cookie value>",
        "csrfAuth": "<csrftoken cookie value>",
        "tokenAuth": "Token <auth key value>",
    }
)

You can perform a regular login using the auth_api member of ApiClient and set the Authorization header using the Token prefix. This way, you’ll be able to obtain API tokens, which can be reused in the future to avoid typing your credentials.

from cvat_sdk.api_client import models

(auth, _) = api_client.auth_api.create_login(
    models.LoginRequest(username=credentials[0], password=credentials[1])
)

assert "sessionid" in api_client.cookies
assert "csrftoken" in api_client.cookies
api_client.set_default_header("Authorization", "Token " + auth.key)

API wrappers

API endpoints are grouped by tags into separate classes in the cvat_sdk.api_client.apis package.

APIs can be accessed as ApiClient object members:

api_client.auth_api.<operation>(...)
api_client.tasks_api.<operation>(...)

And APIs can be instantiated directly like this:

from cvat_sdk.api_client import ApiClient, apis

api_client = ApiClient(...)

auth_api = apis.AuthApi(api_client)
auth_api.<operation>(...)

tasks_api = apis.TasksApi(api_client)
tasks_api.<operation>(...)

For each operation, the API wrapper class has a corresponding <operation>_endpoint member. This member represents the endpoint as a first-class object, which provides metainformation about the endpoint, such as the relative URL of the endpoint, parameter names, types and their placement in the request. It also allows to pass the operation to other functions and invoke it from there.

For a typical server entity like Task, Project, Job etc., the *Api classes provide methods that reflect Create-Read-Update-Delete (CRUD) operations: create, retrieve, list, update, partial_update, delete. The set of available operations depends on the entity type.

You can find the list of the available APIs and their documentation here.

Models

Requests and responses can include data. It can be represented as plain Python data structures and model classes (or models). In CVAT API, model for requests and responses are separated: the request models have the Request suffix in the name, while the response models have no suffix. Models can be found in the cvat_sdk.api_client.models package.

Models can be instantiated like this:

from cvat_sdk.api_client import models

user_model = models.User(...)

Model parameters can be passed as models, or as plain Python data structures. This rule applies recursively, starting from the method parameters. In particular, this means you can pass a dict into a method or into a model constructor, and corresponding fields will be parsed from this data automatically:

task_spec = models.TaskWriteRequest(
    name='example task',
    labels=[
        models.PatchedLabelRequest(
            name="car",
            color="#ff00ff",
            attributes=[
                model.AttributeRequest(
                    name="a",
                    mutable=True,
                    input_type="number",
                    default_value="5",
                    values=["4", "5", "6"]
                )
            ]
        )
    ],
)
api_client.tasks_api.create(task_spec)

Is equivalent to:

api_client.tasks_api.create({
    'name': 'example task',
    "labels": [{
        "name": "car",
        "color": "#ff00ff",
        "attributes": [
            {
                "name": "a",
                "mutable": True,
                "input_type": "number",
                "default_value": "5",
                "values": ["4", "5", "6"]
            }
        ]
    }],
})

You can mix these variants.

Most models provide corresponding interface classes called like I<model name>. They can be used to implement your own classes or describe APIs. They just provide type annotations and descriptions for model fields.

You can export model values to plain Python dicts using the as_dict() method and the cvat_sdk.api_client.model_utils.to_json() function.

You can find the list of the available models and their documentation here.

Sending requests

To send a request to a server endpoint, you need to obtain an instance of the corresponding *Api class. You can find summary about available API classes and supported endpoints here. The *Api instance object allows to send requests to the relevant server endpoints.

By default, all operations return 2 objects: the parsed response data and the response itself.

The first returned value is a model parsed from the response data. If a method does not have any return value, None is always returned as the first value. You can control automatic parsing using the _parse_response method kwarg. When disabled, None is returned.

The second value is the raw response, which can be useful to get response parameters, such as status code, headers, or raw response data. By default, the status code of the response is checked to be positive. In the case of request failure, an exception is raised by default. This behavior can be controlled by the _check_status method kwarg. If the status is not checked, you will need to manually check the response status code and perform actions needed.

A typical endpoint call looks like this:

from cvat_sdk.api_client import ApiClient, apis

with ApiClient(...) as api_client:
    ...
    (data, response) = api_client.tasks_api.list()
    # process the response ...

Operation parameters can be passed as positional or keyword arguments. API methods provide extra common arguments which control invocation logic:

• _parse_response (bool) - Allows to enable and disable response data parsing. When enabled, the response data is parsed into a model or a basic type and returned as the first value. When disabled, the response is not parsed, and None is returned. Can be useful, for instance, if you need to parse data manually, or if you expect an error in the response. Default is True.
• _check_status (bool) - Allows to enable or disable response status checks. When enabled, the response status code is checked to be positive as defined in the HTTP standards. In the case of negative status, an exception is raised. Default is True.
• _validate_inputs (bool): specifies if type checking should be done on the data sent to the server. Default is True.
• _validate_outputs (bool): specifies if type checking should be done on the data received from the server. Default is True.
• _request_timeout (None | int | float | Tuple[int | float, int | float]) - Allows to control timeouts. If one number is provided, it will be the total request timeout. It can also be a tuple with (connection, read) timeouts. Default is None, which means no timeout.
• _content_type (None | str) - Allows to specify the Content-Type header value for the request. Endpoints can support different content types and behave differently depending on the value. For file uploads _content_type="multipart/form-data" must be specified. Read more about file uploads here. Default is application/json.

NOTE: the API is autogenerated. In some cases the server API schema may be incomplete or underspecified. Please report to us all the problems found. A typical problem is that a response data can’t be parsed automatically due to the incorrect schema. In this case, the simplest workaround is to disable response parsing using the _parse_response=False method argument.

You can find many examples of API client usage in REST API tests here.

Organizations

To create resource in the context of an organization, use one of these method arguments:

• org - The unique organization slug
• org_id- The organization id

...
(task, response) = api_client.tasks_api.create(task_spec, org_id=org_id)

Paginated responses

There are several endpoints that allow to request multiple server entities. Typically, these endpoints are called list_.... When there are lots of data, the responses can be paginated to reduce server load. If an endpoint returns paginated data, a single page is returned per request. In some cases all entries need to be retrieved. CVAT doesn’t provide specific API or parameters for this, so the solution is to write a loop to collect and join data from multiple requests. SDK provides an utility function for this at cvat_sdk.core.helpers.get_paginated_collection().

Example:

from cvat_sdk.core.helpers import get_paginated_collection

...
project_tasks = get_paginated_collection(
    api_client.projects_api.list_tasks_endpoint,
    id=project_id,
)

Binary data in requests and responses

At the moment, sending and receiving binary data - such as files - can be difficult via the low-level SDK API. Please use the following recommendations.

Sending data

By default, requests use the application/json content type, which is a text type. However, it’s inefficient to send binary data in this encoding, and the data passed won’t be converted automatically. If you need to send files or other binary data, please specify _content_type="multipart/form-data" in the request parameters:

Example:

(_, response) = api_client.tasks_api.create_data(
    id=42,
    data_request=models.DataRequest(
        client_files=[
            open("image.jpg", 'rb')
        ],
        image_quality=70,
    ),
    _content_type="multipart/form-data", # required
)

Please also note that if there are complex fields in the data (such as nested lists or dicts), they, in turn, cannot be encoded as multipart/form-data, so the recommended solution is to split fields into files and others, and send them in different requests with different content types:

Example:

data = {
    'client_files': [...], # a list of binary files
    'image_quality': ..., # a simple type - int
    'job_file_mapping': [...], # a complex type - list
}

# Initialize uploading
api_client.tasks_api.create_data(
    id=42,
    data_request=models.DataRequest(image_quality=data["image_quality"]),
    upload_start=True,
)

# Upload binary data
api_client.tasks_api.create_data(
    id=42,
    data_request=models.DataRequest(
        client_files=data.pop("client_files"),
        image_quality=data["image_quality"],
    ),
    upload_multiple=True,
    _content_type="multipart/form-data",
)

# Finalize the uploading and send the remaining fields
api_client.tasks_api.create_data(
    id=42,
    data_request=models.DataRequest(**data),
    upload_finish=True,
)

Receiving data

Receiving binary files can also be difficult with the low-level API. To avoid unexpected behavior, it is recommended to specify _parse_response=False in the request parameters. In this case, SDK will not try to parse models from responses, and the response data can be fetched directly from the response:

from time import sleep

# Export a task as a dataset
while True:
    (_, response) = api_client.tasks_api.retrieve_dataset(
        id=42,
        format='COCO 1.0',
        _parse_response=False,
    )
    if response.status == HTTPStatus.CREATED:
        break

    sleep(interval)

(_, response) = api_client.tasks_api.retrieve_dataset(
    id=42,
    format='COCO 1.0',
    action="download",
    _parse_response=False,
)

# Save the resulting file
with open('output_file', 'wb') as output_file:
    output_file.write(response.data)

Different versions of API endpoints

The cloudstorages/id/content REST API endpoint

Warning: The retrieve_content method of cloudstorages_api will be deprecated in 2.5.0 version. We recommend using retrieve_content_v2 method that matches to revised API when using SDK. For backward compatibility, we continue to support the prior interface version until version 2.6.0 is released.

Here you can find the example how to get the bucket content using new method retrieve_content_v2.

from pprint import pprint

from cvat_sdk.api_client import ApiClient, Configuration

next_token = None
files, prefixes = [], []
prefix = ""

with ApiClient(
    configuration=Configuration(host=BASE_URL, username=user, password=password)
) as api_client:
    while True:
        data, response = api_client.cloudstorages_api.retrieve_content_v2(
            cloud_storage_id,
            **({"prefix": prefix} if prefix else {}),
            **({"next_token": next_token} if next_token else {}),
        )
        # the data will have the following structure:
        # {'content': [
        #     {'mime_type': <image|video|archive|pdf|DIR>, 'name': <name>, 'type': <REG|DIR>},
        # ],
        # 'next': <next_token_string|None>}
        files.extend(
            [
                prefix + f["name"]
                for f in data["content"]
                if str(f["type"]) == "REG"
            ]
        )
        prefixes.extend(
            [
                prefix + f["name"]
                for f in data["content"]
                if str(f["type"]) == "DIR"
            ]
        )
        next_token = data["next"]
        if next_token:
            continue
        if not len(prefixes):
            break
        prefix = f"{prefixes.pop()}/"
    pprint(files) # ['sub/image_1.jpg', 'image_2.jpg']