cvat/site/content/en/docs/api_sdk/sdk/lowlevel-api.md

735 lines
24 KiB
Markdown
Raw Permalink Normal View History

2025-09-16 01:19:40 +00:00
---
title: 'Low-level API'
linkTitle: 'Low-level API'
weight: 3
description: ''
---
## 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.
```python
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.
(result, 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):
request_details, response = api_client.requests_api.retrieve(result.rq_id)
status, message = request_details.status, request_details.message
if status.value in {'finished', 'failed'}:
break
sleep(0.1)
assert status.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:
```python
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](#api-wrappers) about API wrappers below.
Typically, the first thing you do with `ApiClient` is log in.
[Read more](#authentication) about authentication options below.
## Authentication
CVAT supports 3 authentication options:
- Basic authentication, with a username and a password
- Session authentication, with a session ID and a CSRF token
- Token authentication, with an API key (deprecated)
Token authentication requires an API key, which can be obtained after logging in
via the `/api/auth/login` endpoint using the basic authentication credentials.
Session authentication requires a session ID and a CSRF token, which can be obtained after
logging in via the `/api/auth/login` endpoint using the basic authentication credentials.
Authentication credentials for an `ApiClient` instance can be specified in a `Configuration` object:
{{< tabpane text=true >}}
{{%tab header="Basic authentication" %}}
```python
configuration = Configuration(
username='YOUR_USERNAME',
password='YOUR_PASSWORD',
...
)
with ApiClient(configuration) as api_client:
...
```
{{% /tab %}}
{{%tab header="Session authentication" %}}
```python
configuration = Configuration(
api_key={
"sessionAuth": "<sessionid cookie value>",
"csrfAuth": "<csrftoken cookie value>",
},
...
)
with ApiClient(configuration) as api_client:
...
```
{{% /tab %}}
{{%tab header="Token authentication (deprecated)" %}}
{{% alert title="Warning" color="warning" %}}
This authentication option is deprecated and will be removed in future.
{{% /alert %}}
```python
configuration = Configuration(
api_key={
"tokenAuth": "Token <api key value>",
},
...
)
with ApiClient(configuration) as api_client:
...
```
{{% /tab %}}
{{< /tabpane >}}
Session authentication and token authentication tokens can be received by logging in
using the `ApiClient.auth_api.create_login()` function. Then, the authentication keys
can be set in the `ApiClient` instance.
{{< tabpane text=true >}}
{{%tab header="Session authentication" %}}
```python
from cvat_sdk.api_client import models
(auth, _) = api_client.auth_api.create_login(
models.LoginSerializerExRequest(username="username", password="password")
)
# Set up required headers
assert "sessionid" in api_client.cookies # managed by ApiClient automatically
api_client.set_default_header("X-CSRFToken", api_client.cookies["csrftoken"].value)
api_client.set_default_header("Origin", api_client.build_origin_header())
```
{{% /tab %}}
{{%tab header="Token authentication (deprecated)" %}}
{{% alert title="Warning" color="warning" %}}
This authentication option is deprecated and will be removed in future.
{{% /alert %}}
```python
from cvat_sdk.api_client import models
(auth, _) = api_client.auth_api.create_login(
models.LoginSerializerExRequest(username="username", password="password")
)
api_client.set_default_header("Authorization", "Token " + auth.key)
```
{{% /tab %}}
{{< /tabpane >}}
## 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:
```python
api_client.auth_api.<operation>(...)
api_client.tasks_api.<operation>(...)
```
And APIs can be instantiated directly like this:
```python
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](../reference/apis/).
## 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:
```python
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:
```python
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:
```python
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 `to_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](../reference/models/).
## 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](../reference/apis). 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:
```python
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](https://en.wikipedia.org/wiki/List_of_HTTP_status_codes).
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](#sending-data). Default is `application/json`.
{{% alert title="Note" color="primary" %}}
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.
{{% /alert %}}
You can find many examples of API client usage in REST API tests [here](https://github.com/cvat-ai/cvat/tree/develop/tests/python).
### 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
```python
...
(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:
```python
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:
```python
(result, 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:
```python
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:
```python
import json
from http import HTTPStatus
from time import sleep
from urllib.parse import parse_qsl, urlparse
from cvat_sdk.api_client import ApiClient, Configuration, models
interval = 1
with ApiClient(
configuration=Configuration(host="<cvat_host>", username="<username>", password="<password>")
) as api_client:
# Initiate the process to export a task as a dataset
(_, response) = api_client.tasks_api.create_dataset_export(
id=task_id,
format="COCO 1.0",
save_images=True,
_parse_response=False,
)
assert response.status == HTTPStatus.ACCEPTED
# Obtain the background request ID from the server response
rq_id = json.loads(response.data).get("rq_id")
assert rq_id, "The rq_id parameter was not found in the server response"
# Check the status of the background process
while True:
(background_request, response) = api_client.requests_api.retrieve(rq_id)
assert response.status == HTTPStatus.OK
process_status = background_request.status.value
if process_status in (
models.RequestStatus.allowed_values[("value",)]["FINISHED"],
models.RequestStatus.allowed_values[("value",)]["FAILED"],
):
break
sleep(interval)
if process_status != models.RequestStatus.allowed_values[("value",)]["FINISHED"]:
exception_msg = f"Export failed with status: {process_status}"
if background_request.message:
exception_msg += f". Details: {background_request.message}"
assert False, exception_msg
# Download a prepared file
result_url = background_request.result_url
assert result_url, "No 'result_url' in the server response"
parsed_result_url = urlparse(result_url)
query_params = parse_qsl(parsed_result_url.query)
_, response = api_client.call_api(
parsed_result_url.path,
method="GET",
query_params=query_params,
auth_settings=api_client.configuration.auth_settings(),
_parse_response=False,
)
# Save the resulting file
with open("output_file.zip", "wb") as output_file:
while (chunk := response.read(8192)):
output_file.write(chunk)
```
### Different versions of API endpoints
#### The cloudstorages/id/content REST API endpoint
{{% alert title="Warning" color="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.
{{% /alert %}}
Here you can find the example how to get the bucket content using new method `retrieve_content_v2`.
```python
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']
```
### Sending custom requests
Sometimes you might need sending a custom request while using the `ApiClient`.
Typically, it's desirable to make this request using the same configuration
and authentication parameters as regular requests sent via the `ApiClient` instance.
One particularly useful example for this is dataset export. When exporting a dataset,
you receive a download URL from the server after the file is prepared
(see example in [receiving data](#receiving-data)). This URL requires authentication,
but it can't be made via the regular `ApiClient` endpoint methods.
There are several options to make such a custom request via an `ApiClient` instance:
- use `api_client.call_api()`
- use `api_client.request()`
Alternatively, it's also possible to make a request using a custom backend for requests, while
keeping the existing authentication of the `ApiClient`.
{{< tabpane text=true >}}
{{% tab header="Using api_client.call_api()" %}}
This option provides higher-level interface and allows better integration with other
`ApiClient`-related interfaces, such as `Endpoint`.
```python
from urllib.parse import parse_qsl, urlparse
with ApiClient(...) as api_client:
parsed_url = urlparse("<custom URL>")
query_params = parse_qsl(parsed_url.query)
_, response = api_client.call_api(
parsed_url.path,
method="GET",
query_params=query_params,
_parse_response=False,
)
# process response.data ...
```
{{% /tab %}}
{{% tab header="Using api_client.request()" %}}
This option provides a low-level interface and allows more customization. It can be useful
if you need to reuse the existing connection configuration of the `ApiClient`
instance, such as connection pool, timeouts, etc. In order to keep the existing
authentication parameters of the `ApiClient` instance, you
can use the functions `api_client.get_common_headers()` and `api_client.update_params_for_auth()`.
```python
with ApiClient(...) as api_client:
headers = api_client.get_common_headers()
api_client.update_params_for_auth(headers=headers, queries=[], method="GET")
response = api_client.request(
"GET",
"<custom URL>",
headers=headers,
_parse_response=False,
)
# process response.data ...
```
{{% /tab %}}
{{% tab header="Using a custom backend" %}}
It is possible make a custom request using your own backend - for example, using the `requests`
library. In order to keep the existing authentication parameters of the `ApiClient` instance, you
can use the functions `api_client.get_common_headers()` and `api_client.update_params_for_auth()`.
```python
import requests
with ApiClient(...) as api_client:
headers = api_client.get_common_headers()
query_params = []
api_client.update_params_for_auth(headers=headers, queries=query_params, method="GET")
response = requests.get("<custom URL>", params=query_params, headers=headers)
# process the response ...
```
{{% /tab %}}
{{< /tabpane >}}