Repository: cvat-ai/cvat
Pull request: Tus for task annotations import
<!---
Copyright (C) 2020-2022 Intel Corporation

SPDX-License-Identifier: MIT
-->

<!-- Raised an issue to propose your change (https://github.com/opencv/cvat/issues).
It helps to avoid duplication of efforts from multiple independent contributors.
Discuss your ideas with maintainers to be sure that changes will be approved and merged.
Read the [CONTRIBUTION](https://github.com/opencv/cvat/blob/develop/CONTRIBUTING.md)
guide. -->

<!-- Provide a general summary of your changes in the Title above -->

### Motivation and context
<!-- Why is this change required? What problem does it solve? If it fixes an open
issue, please link to the issue here. Describe your changes in detail, add
screenshots. -->
Continue TUS integration to provide chunk upload for large annotation files in Tasks and Jobs
Resolved #964 
### How has this been tested?
<!-- Please describe in detail how you tested your changes.
Include details of your testing environment, and the tests you ran to
see how your change affects other areas of the code, etc. -->

### Checklist
<!-- Go over all the following points, and put an `x` in all the boxes that apply.
If an item isn't applicable by a reason then ~~explicitly strikethrough~~ the whole
line. If you don't do that github will show an incorrect process for the pull request.
If you're unsure about any of these, don't hesitate to ask. We're here to help! -->
- [x] I submit my changes into the `develop` branch
- [x] I have added a description of my changes into [CHANGELOG](https://github.com/opencv/cvat/blob/develop/CHANGELOG.md) file
- [ ] I have updated the [documentation](
  https://github.com/opencv/cvat/blob/develop/README.md#documentation) accordingly
- [ ] I have added tests to cover my changes
- [x] I have linked related issues ([read github docs](
  https://help.github.com/en/github/managing-your-work-on-github/linking-a-pull-request-to-an-issue#linking-a-pull-request-to-an-issue-using-a-keyword))
- [x] I have increased versions of npm packages if it is necessary ([cvat-canvas](https://github.com/opencv/cvat/tree/develop/cvat-canvas#versioning),
[cvat-core](https://github.com/opencv/cvat/tree/develop/cvat-core#versioning), [cvat-data](https://github.com/opencv/cvat/tree/develop/cvat-data#versioning) and [cvat-ui](https://github.com/opencv/cvat/tree/develop/cvat-ui#versioning))

### License

- [x] I submit _my code changes_ under the same [MIT License](
  https://github.com/opencv/cvat/blob/develop/LICENSE) that covers the project.
  Feel free to contact the maintainers if that's a concern.
- [x] I have updated the license header for each file (see an example below)

```python
# Copyright (C) 2022 Intel Corporation
#
# SPDX-License-Identifier: MIT
```


Problem statement:
Timeout when uploading a big file with annotations
When I try upload 100 MB xml file annotation, I get this error on log and "Failed to load resource: the server responded with a status of 504 (Gateway Timeout)" error on console.

How can I increase the timeout of the upload annotation?

Hints from issue discussion:
Probably it will be necessary to do the following changes:
- Optimize upload
- Change REST API because the operation can be heavy
Thanks for the replay!
Meanwhile I wander if there is possibility this parameter  "Timeout 60" or "request-timeout=60" in the httpd.conf file.
I was trying to change this file manually, but the change didn't effect. Those parameters are constants and I didn't find which process creates this file.
Do you know which process creates this httpd.conf file, or how to change those parameters?
is it possible to upload the annotation (250 MB) to the server and run some CLI instead? 
@gitunit 
Yes, it is possible.
@alontrais good to know. how would you do it?
Related #673 
@klakhov , it is something which you need to fix in your next PR.

https://github.com/openvinotoolkit/cvat/pull/3692

Commit message: added tus for jobs annotations import

Patch to review:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 62372d0614d0..16d0551b6bb2 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -33,7 +33,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Basic page with jobs list, basic filtration to this list (<https://github.com/openvinotoolkit/cvat/pull/4258>)
 - Added OpenCV.js TrackerMIL as tracking tool (<https://github.com/openvinotoolkit/cvat/pull/4200>)
 - Ability to continue working from the latest frame where an annotator was before (<https://github.com/openvinotoolkit/cvat/pull/4297>)
-
+- Task annotations importing via chunk uploads (<https://github.com/openvinotoolkit/cvat/pull/4327>)
 
 ### Changed
 - Users don't have access to a task object anymore if they are assigneed only on some jobs of the task (<https://github.com/openvinotoolkit/cvat/pull/3788>)
diff --git a/cvat-core/package-lock.json b/cvat-core/package-lock.json
index 3c3f7a6aa954..6d32e808cc9a 100644
--- a/cvat-core/package-lock.json
+++ b/cvat-core/package-lock.json
@@ -1,12 +1,12 @@
 {
   "name": "cvat-core",
-  "version": "4.2.0",
+  "version": "4.2.1",
   "lockfileVersion": 2,
   "requires": true,
   "packages": {
     "": {
       "name": "cvat-core",
-      "version": "4.2.0",
+      "version": "4.2.1",
       "license": "MIT",
       "dependencies": {
         "axios": "^0.21.4",
diff --git a/cvat-core/package.json b/cvat-core/package.json
index 106426cb2cfe..098ddd8a2e13 100644
--- a/cvat-core/package.json
+++ b/cvat-core/package.json
@@ -1,6 +1,6 @@
 {
   "name": "cvat-core",
-  "version": "4.2.0",
+  "version": "4.2.1",
   "description": "Part of Computer Vision Tool which presents an interface for client-side integration",
   "main": "babel.config.js",
   "scripts": {
diff --git a/cvat-core/src/server-proxy.js b/cvat-core/src/server-proxy.js
index ca46f12d5f79..146131dbf4e4 100644
--- a/cvat-core/src/server-proxy.js
+++ b/cvat-core/src/server-proxy.js
@@ -45,6 +45,49 @@
         });
     }
 
+    async function chunkUpload(file, uploadConfig) {
+        const params = enableOrganization();
+        const {
+            endpoint, chunkSize, totalSize, onUpdate,
+        } = uploadConfig;
+        let { totalSentSize } = uploadConfig;
+        return new Promise((resolve, reject) => {
+            const upload = new tus.Upload(file, {
+                endpoint,
+                metadata: {
+                    filename: file.name,
+                    filetype: file.type,
+                },
+                headers: {
+                    Authorization: Axios.defaults.headers.common.Authorization,
+                },
+                chunkSize,
+                retryDelays: null,
+                onError(error) {
+                    reject(error);
+                },
+                onBeforeRequest(req) {
+                    const xhr = req.getUnderlyingObject();
+                    const { org } = params;
+                    req.setHeader('X-Organization', org);
+                    xhr.withCredentials = true;
+                },
+                onProgress(bytesUploaded) {
+                    if (Number.isInteger(totalSentSize) && Number.isInteger(totalSize)) {
+                        const currentUploadedSize = totalSentSize + bytesUploaded;
+                        const percentage = currentUploadedSize / totalSize;
+                        if (onUpdate) onUpdate(percentage);
+                    }
+                },
+                onSuccess() {
+                    if (totalSentSize) totalSentSize += file.size;
+                    resolve(totalSentSize);
+                },
+            });
+            upload.start();
+        });
+    }
+
     function generateError(errorData) {
         if (errorData.response) {
             const message = `${errorData.message}. ${JSON.stringify(errorData.response.data) || ''}.`;
@@ -816,42 +859,6 @@
 
                 onUpdate('The data are being uploaded to the server..', null);
 
-                async function chunkUpload(taskId, file) {
-                    return new Promise((resolve, reject) => {
-                        const upload = new tus.Upload(file, {
-                            endpoint: `${origin}${backendAPI}/tasks/${taskId}/data/`,
-                            metadata: {
-                                filename: file.name,
-                                filetype: file.type,
-                            },
-                            headers: {
-                                Authorization: `Token ${store.get('token')}`,
-                            },
-                            chunkSize,
-                            retryDelays: null,
-                            onError(error) {
-                                reject(error);
-                            },
-                            onBeforeRequest(req) {
-                                const xhr = req.getUnderlyingObject();
-                                const { org } = params;
-                                req.setHeader('X-Organization', org);
-                                xhr.withCredentials = true;
-                            },
-                            onProgress(bytesUploaded) {
-                                const currentUploadedSize = totalSentSize + bytesUploaded;
-                                const percentage = currentUploadedSize / totalSize;
-                                onUpdate('The data are being uploaded to the server', percentage);
-                            },
-                            onSuccess() {
-                                totalSentSize += file.size;
-                                resolve();
-                            },
-                        });
-                        upload.start();
-                    });
-                }
-
                 async function bulkUpload(taskId, files) {
                     const fileBulks = files.reduce((fileGroups, file) => {
                         const lastBulk = fileGroups[fileGroups.length - 1];
@@ -891,8 +898,17 @@
                             proxy: config.proxy,
                             headers: { 'Upload-Start': true },
                         });
+                    const uploadConfig = {
+                        endpoint: `${origin}${backendAPI}/tasks/${response.data.id}/data/`,
+                        onUpdate: (percentage) => {
+                            onUpdate('The data are being uploaded to the server', percentage);
+                        },
+                        chunkSize,
+                        totalSize,
+                        totalSentSize,
+                    };
                     for (const file of chunkFiles) {
-                        await chunkUpload(response.data.id, file);
+                        uploadConfig.totalSentSize += await chunkUpload(file, uploadConfig);
                     }
                     if (bulkFiles.length > 0) {
                         await bulkUpload(response.data.id, bulkFiles);
@@ -1219,34 +1235,54 @@
                 const params = {
                     ...enableOrganization(),
                     format,
+                    filename: file.name,
                 };
-                let annotationData = new FormData();
-                annotationData.append('annotation_file', file);
+                const chunkSize = config.uploadChunkSize * 1024 * 1024;
 
-                return new Promise((resolve, reject) => {
-                    async function request() {
-                        try {
-                            const response = await Axios.put(
-                                `${backendAPI}/${session}s/${id}/annotations`,
-                                annotationData,
-                                {
-                                    params,
-                                    proxy: config.proxy,
-                                },
-                            );
-                            if (response.status === 202) {
-                                annotationData = new FormData();
-                                setTimeout(request, 3000);
-                            } else {
-                                resolve();
+                const uploadConfig = {
+                    chunkSize,
+                    endpoint: `${origin}${backendAPI}/${session}s/${id}/annotations/`,
+                };
+                try {
+                    await Axios.post(`${backendAPI}/${session}s/${id}/annotations`,
+                        new FormData(), {
+                            params,
+                            proxy: config.proxy,
+                            headers: { 'Upload-Start': true },
+                        });
+                    await chunkUpload(file, uploadConfig);
+                    await Axios.post(`${backendAPI}/${session}s/${id}/annotations`,
+                        new FormData(), {
+                            params,
+                            proxy: config.proxy,
+                            headers: { 'Upload-Finish': true },
+                        });
+                    return new Promise((resolve, reject) => {
+                        async function requestStatus() {
+                            try {
+                                const response = await Axios.put(
+                                    `${backendAPI}/${session}s/${id}/annotations`,
+                                    new FormData(),
+                                    {
+                                        params,
+                                        proxy: config.proxy,
+                                    },
+                                );
+                                if (response.status === 202) {
+                                    setTimeout(requestStatus, 3000);
+                                } else {
+                                    resolve();
+                                }
+                            } catch (errorData) {
+                                reject(generateError(errorData));
                             }
-                        } catch (errorData) {
-                            reject(generateError(errorData));
                         }
-                    }
-
-                    setTimeout(request);
-                });
+                        setTimeout(requestStatus);
+                    });
+                } catch (errorData) {
+                    generateError(errorData);
+                    return null;
+                }
             }
 
             // Session is 'task' or 'job'
diff --git a/cvat/apps/engine/mixins.py b/cvat/apps/engine/mixins.py
index 9cb1a4808a2d..1f48d233fda2 100644
--- a/cvat/apps/engine/mixins.py
+++ b/cvat/apps/engine/mixins.py
@@ -26,6 +26,7 @@ def __init__(self, file_id, upload_dir):
         self.offset = cache.get("tus-uploads/{}/offset".format(file_id))
 
     def init_file(self):
+        os.makedirs(self.upload_dir, exist_ok=True)
         file_path = os.path.join(self.upload_dir, self.file_id)
         with open(file_path, 'wb') as file:
             file.seek(self.file_size - 1)
@@ -101,6 +102,7 @@ class UploadMixin(object):
         'Cache-Control': 'no-store'
     }
     _file_id_regex = r'(?P<file_id>\b[0-9a-f]{8}\b-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-\b[0-9a-f]{12}\b)'
+    upload_url = r'((data/)|(annotations/))'
 
     def _tus_response(self, status, data=None, extra_headers=None):
         response = Response(data, status)
@@ -126,33 +128,39 @@ def _get_metadata(self, request):
                     metadata[splited_metadata[0]] = ""
         return metadata
 
-    def upload_data(self, request):
+    def _get_chunk_data_type(self, request):
+        path_elements = request.path.split('/')
+        # When we use TUS chunk upload we consider path structure as <protocol>://<domain>/.../<data type>/<file id>
+        chunk_data_type = path_elements[-2]
+        return chunk_data_type
+
+    def upload_data(self, request, data_type):
         tus_request = request.headers.get('Upload-Length', None) is not None or request.method == 'OPTIONS'
         bulk_file_upload = request.headers.get('Upload-Multiple', None) is not None
         start_upload = request.headers.get('Upload-Start', None) is not None
         finish_upload = request.headers.get('Upload-Finish', None) is not None
         one_request_upload = start_upload and finish_upload
         if one_request_upload or finish_upload:
-            return self.upload_finished(request)
+            return self.upload_finished(request, data_type)
         elif start_upload:
             return Response(status=status.HTTP_202_ACCEPTED)
         elif tus_request:
-            return self.init_tus_upload(request)
+            return self.init_tus_upload(request, data_type)
         elif bulk_file_upload:
-            return self.append(request)
+            return self.append(request, data_type)
         else: # backward compatibility case - no upload headers were found
-            return self.upload_finished(request)
+            return self.upload_finished(request, data_type)
 
-    def init_tus_upload(self, request):
+    def init_tus_upload(self, request, data_type):
         if request.method == 'OPTIONS':
             return self._tus_response(status=status.HTTP_204)
         else:
-            if not self.can_upload():
+            if not self.can_upload(data_type):
                 return self._tus_response(data='Adding more data is not allowed',
                     status=status.HTTP_400_BAD_REQUEST)
             metadata = self._get_metadata(request)
             filename = metadata.get('filename', '')
-            if not self.validate_filename(filename):
+            if not self.validate_filename(filename, data_type):
                 return self._tus_response(status=status.HTTP_400_BAD_REQUEST,
                     data="File name {} is not allowed".format(filename))
 
@@ -161,7 +169,7 @@ def init_tus_upload(self, request):
             if message_id:
                 metadata["message_id"] = base64.b64decode(message_id)
 
-            file_exists = os.path.lexists(os.path.join(self.get_upload_dir(), filename))
+            file_exists = os.path.lexists(os.path.join(self.get_upload_dir(data_type), filename))
             if file_exists:
                 return self._tus_response(status=status.HTTP_409_CONFLICT,
                     data="File with same name already exists")
@@ -171,24 +179,24 @@ def init_tus_upload(self, request):
                 return self._tus_response(status=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE,
                     data="File size exceeds max limit of {} bytes".format(self._tus_max_file_size))
 
-            tus_file = TusFile.create_file(metadata, file_size, self.get_upload_dir())
+            tus_file = TusFile.create_file(metadata, file_size, self.get_upload_dir(data_type))
 
             return self._tus_response(
                 status=status.HTTP_201_CREATED,
                 extra_headers={'Location': '{}{}'.format(request.build_absolute_uri(), tus_file.file_id)})
 
-    @action(detail=True, methods=['HEAD', 'PATCH'], url_path=r'data/'+_file_id_regex)
+    @action(detail=True, methods=['HEAD', 'PATCH'], url_path=upload_url+_file_id_regex)
     def append_tus_chunk(self, request, pk, file_id):
-        self.get_object() # call check_object_permissions as well
+        data_type = self._get_chunk_data_type(request)
         if request.method == 'HEAD':
-            tus_file = TusFile.get_tusfile(str(file_id), self.get_upload_dir())
+            tus_file = TusFile.get_tusfile(str(file_id), self.get_upload_dir(data_type))
             if tus_file:
                 return self._tus_response(status=status.HTTP_200_OK, extra_headers={
                                'Upload-Offset': tus_file.offset,
                                'Upload-Length': tus_file.file_size})
             return self._tus_response(status=status.HTTP_404_NOT_FOUND)
         else:
-            tus_file = TusFile.get_tusfile(str(file_id), self.get_upload_dir())
+            tus_file = TusFile.get_tusfile(str(file_id), self.get_upload_dir(data_type))
             chunk = TusChunk(request)
 
             if chunk.offset != tus_file.offset:
@@ -206,17 +214,15 @@ def append_tus_chunk(self, request, pk, file_id):
             return self._tus_response(status=status.HTTP_204_NO_CONTENT,
                                     extra_headers={'Upload-Offset': tus_file.offset})
 
-    def validate_filename(self, filename):
-        upload_dir = self.get_upload_dir()
+    def validate_filename(self, filename, data_type):
+        upload_dir = self.get_upload_dir(data_type)
         file_path = os.path.join(upload_dir, filename)
         return os.path.commonprefix((os.path.realpath(file_path), upload_dir)) == upload_dir
 
-    def can_upload(self):
-        db_model = self.get_object()
-        model_data = db_model.data
-        return model_data.size == 0
+    def can_upload(self, data_type):
+        return True
 
-    def get_upload_dir(self):
+    def get_upload_dir(self, data_type):
         db_model = self.get_object()
         return db_model.data.get_upload_dirname()
 
@@ -227,18 +233,18 @@ def get_request_client_files(self, request):
         data = {k: v for k, v in serializer.validated_data.items()}
         return data.get('client_files', None);
 
-    def append(self, request):
-        if not self.can_upload():
+    def append(self, request, data_type):
+        if not self.can_upload(data_type):
             return Response(data='Adding more data is not allowed',
                 status=status.HTTP_400_BAD_REQUEST)
         client_files = self.get_request_client_files(request)
         if client_files:
-            upload_dir = self.get_upload_dir()
+            upload_dir = self.get_upload_dir(data_type)
             for client_file in client_files:
                 with open(os.path.join(upload_dir, client_file['file'].name), 'ab+') as destination:
                     destination.write(client_file['file'].read())
         return Response(status=status.HTTP_200_OK)
 
     # override this to do stuff after upload
-    def upload_finished(self, request):
+    def upload_finished(self, request, data_type):
         raise NotImplementedError('You need to implement upload_finished in UploadMixin')
diff --git a/cvat/apps/engine/models.py b/cvat/apps/engine/models.py
index ede2bf717282..622e0489589a 100644
--- a/cvat/apps/engine/models.py
+++ b/cvat/apps/engine/models.py
@@ -304,6 +304,15 @@ def get_log_path(self):
     def get_task_artifacts_dirname(self):
         return os.path.join(self.get_task_dirname(), 'artifacts')
 
+    def get_tmp_dirname(self):
+        return os.path.join(self.get_task_dirname(), "tmp")
+
+    def get_tmp_file(self, filename):
+        tmp_dir = self.get_tmp_dirname()
+        if filename in os.listdir(tmp_dir) and os.path.isfile(os.path.join(tmp_dir, filename)):
+            return os.path.join(tmp_dir, filename)
+        return None
+
     def __str__(self):
         return self.name
 
diff --git a/cvat/apps/engine/views.py b/cvat/apps/engine/views.py
index e6841760be63..f2c8e44c5b4a 100644
--- a/cvat/apps/engine/views.py
+++ b/cvat/apps/engine/views.py
@@ -685,39 +685,75 @@ def jobs(self, request, pk):
 
         return Response(serializer.data)
 
-    def upload_finished(self, request):
+    # UploadMixin method
+    def can_upload(self, data_type):
+        if data_type == 'data':
+            db_model = self.get_object()
+            model_data = db_model.data
+            return model_data.size == 0
+        return True
+
+    # UploadMixin method
+    def get_upload_dir(self, data_type):
+        db_model = self.get_object()
+        if data_type == 'annotations':
+            return db_model.get_tmp_dirname()
+        elif data_type == 'data':
+            return db_model.data.get_upload_dirname()
+        return ""
+
+    # UploadMixin method
+    def upload_finished(self, request, data_type):
         db_task = self.get_object() # call check_object_permissions as well
         task_data = db_task.data
-        serializer = DataSerializer(task_data, data=request.data)
-        serializer.is_valid(raise_exception=True)
-        data = dict(serializer.validated_data.items())
-        uploaded_files = task_data.get_uploaded_files()
-        uploaded_files.extend(data.get('client_files'))
-        serializer.validated_data.update({'client_files': uploaded_files})
-
-        db_data = serializer.save()
-        db_task.data = db_data
-        db_task.save()
-        data = {k: v for k, v in serializer.data.items()}
-
-        data['use_zip_chunks'] = serializer.validated_data['use_zip_chunks']
-        data['use_cache'] = serializer.validated_data['use_cache']
-        data['copy_data'] = serializer.validated_data['copy_data']
-        if data['use_cache']:
-            db_task.data.storage_method = StorageMethodChoice.CACHE
-            db_task.data.save(update_fields=['storage_method'])
-        if data['server_files'] and not data.get('copy_data'):
-            db_task.data.storage = StorageChoice.SHARE
-            db_task.data.save(update_fields=['storage'])
-        if db_data.cloud_storage:
-            db_task.data.storage = StorageChoice.CLOUD_STORAGE
-            db_task.data.save(update_fields=['storage'])
-            # if the value of stop_frame is 0, then inside the function we cannot know
-            # the value specified by the user or it's default value from the database
-        if 'stop_frame' not in serializer.validated_data:
-            data['stop_frame'] = None
-        task.create(db_task.id, data)
-        return Response(serializer.data, status=status.HTTP_202_ACCEPTED)
+        if data_type == 'annotations':
+            format_name = request.query_params.get("format", "")
+            filename = request.query_params.get("filename", "")
+            annotation_file = db_task.get_tmp_file(filename)
+            if not annotation_file:
+                return Response(data='No such file were uploaded',
+                        status=status.HTTP_400_BAD_REQUEST)
+            return _import_annotations(
+                        request=request,
+                        filename=annotation_file,
+                        rq_id="{}@/api/tasks/{}/annotations/upload".format(request.user, db_task.pk),
+                        rq_func=dm.task.import_task_annotations,
+                        pk=db_task.pk,
+                        format_name=format_name,
+                    )
+        elif data_type == 'data':
+            serializer = DataSerializer(task_data, data=request.data)
+            serializer.is_valid(raise_exception=True)
+            data = dict(serializer.validated_data.items())
+            uploaded_files = task_data.get_uploaded_files()
+            uploaded_files.extend(data.get('client_files'))
+            serializer.validated_data.update({'client_files': uploaded_files})
+
+            db_data = serializer.save()
+            db_task.data = db_data
+            db_task.save()
+            data = {k: v for k, v in serializer.data.items()}
+
+            data['use_zip_chunks'] = serializer.validated_data['use_zip_chunks']
+            data['use_cache'] = serializer.validated_data['use_cache']
+            data['copy_data'] = serializer.validated_data['copy_data']
+            if data['use_cache']:
+                db_task.data.storage_method = StorageMethodChoice.CACHE
+                db_task.data.save(update_fields=['storage_method'])
+            if data['server_files'] and not data.get('copy_data'):
+                db_task.data.storage = StorageChoice.SHARE
+                db_task.data.save(update_fields=['storage'])
+            if db_data.cloud_storage:
+                db_task.data.storage = StorageChoice.CLOUD_STORAGE
+                db_task.data.save(update_fields=['storage'])
+                # if the value of stop_frame is 0, then inside the function we cannot know
+                # the value specified by the user or it's default value from the database
+            if 'stop_frame' not in serializer.validated_data:
+                data['stop_frame'] = None
+            task.create(db_task.id, data)
+            return Response(serializer.data, status=status.HTTP_202_ACCEPTED)
+        return Response(data='Unknown data type upload was finished',
+                        status=status.HTTP_400_BAD_REQUEST)
 
     @extend_schema(methods=['POST'],
         summary='Method permanently attaches images or video to a task. Supports tus uploads, see more https://tus.io/',
@@ -760,7 +796,7 @@ def data(self, request, pk):
             elif task_data.size != 0:
                 return Response(data='Adding more data is not supported',
                     status=status.HTTP_400_BAD_REQUEST)
-            return self.upload_data(request)
+            return self.upload_data(request, 'data')
 
         else:
             data_type = request.query_params.get('type', None)
@@ -808,7 +844,7 @@ def data(self, request, pk):
         responses={
             '204': OpenApiResponse(description='The annotation has been deleted'),
         }, tags=['tasks'], versions=['2.0'])
-    @action(detail=True, methods=['GET', 'DELETE', 'PUT', 'PATCH'],
+    @action(detail=True, methods=['GET', 'DELETE', 'PUT', 'PATCH', 'POST', 'OPTIONS'], url_path=r'annotations/?$',
         serializer_class=LabeledDataSerializer)
     def annotations(self, request, pk):
         db_task = self.get_object() # force to call check_object_permissions
@@ -828,6 +864,8 @@ def annotations(self, request, pk):
                 serializer = LabeledDataSerializer(data=data)
                 if serializer.is_valid(raise_exception=True):
                     return Response(serializer.data)
+        elif request.method == 'POST' or request.method == 'OPTIONS':
+            return self.upload_data(request, 'annotations')
         elif request.method == 'PUT':
             format_name = request.query_params.get('format')
             if format_name:
@@ -989,7 +1027,7 @@ class Meta:
        '200': JobWriteSerializer,
    }, tags=['jobs'], versions=['2.0']))
 class JobViewSet(viewsets.GenericViewSet, mixins.ListModelMixin,
-    mixins.RetrieveModelMixin, mixins.UpdateModelMixin):
+    mixins.RetrieveModelMixin, mixins.UpdateModelMixin, UploadMixin):
     queryset = Job.objects.all().order_by('id')
     filterset_class = JobFilter
     iam_organization_field = 'segment__task__organization'
@@ -1008,6 +1046,34 @@ def get_serializer_class(self):
         else:
             return JobWriteSerializer
 
+    # UploadMixin method
+    def get_upload_dir(self, data_type):
+        db_job = self.get_object()
+        task = db_job.segment.task
+        return task.get_tmp_dirname()
+
+    # UploadMixin method
+    def upload_finished(self, request, data_type):
+        db_job = self.get_object()
+        task = db_job.segment.task
+        if data_type == 'annotations':
+            format_name = request.query_params.get("format", "")
+            filename = request.query_params.get("filename", "")
+            annotation_file = task.get_tmp_file(filename)
+            if not annotation_file:
+                return Response(data='No such file were uploaded',
+                        status=status.HTTP_400_BAD_REQUEST)
+            return _import_annotations(
+                        request=request,
+                        filename=annotation_file,
+                        rq_id="{}@/api/jobs/{}/annotations/upload".format(request.user, db_job.pk),
+                        rq_func=dm.task.import_job_annotations,
+                        pk=db_job.pk,
+                        format_name=format_name,
+                    )
+        return Response(data='Unknown data type upload was finished',
+                        status=status.HTTP_400_BAD_REQUEST)
+
     @extend_schema(methods=['GET'], summary='Method returns annotations for a specific job',
         responses={
             '200': LabeledDataSerializer(many=True),
@@ -1031,13 +1097,15 @@ def get_serializer_class(self):
         responses={
             '204': OpenApiResponse(description='The annotation has been deleted'),
         }, tags=['jobs'], versions=['2.0'])
-    @action(detail=True, methods=['GET', 'DELETE', 'PUT', 'PATCH'],
+    @action(detail=True, methods=['GET', 'DELETE', 'PUT', 'PATCH', 'POST', 'OPTIONS'], url_path=r'annotations/?$',
         serializer_class=LabeledDataSerializer)
     def annotations(self, request, pk):
         self.get_object() # force to call check_object_permissions
         if request.method == 'GET':
             data = dm.task.get_job_data(pk)
             return Response(data)
+        elif request.method == 'POST' or request.method == 'OPTIONS':
+            return self.upload_data(request, 'annotations')
         elif request.method == 'PUT':
             format_name = request.query_params.get("format", "")
             if format_name:
@@ -1612,7 +1680,7 @@ def rq_handler(job, exc_type, exc_value, tb):
 
     return True
 
-def _import_annotations(request, rq_id, rq_func, pk, format_name):
+def _import_annotations(request, rq_id, rq_func, pk, format_name, filename=None):
     format_desc = {f.DISPLAY_NAME: f
         for f in dm.views.get_import_formats()}.get(format_name)
     if format_desc is None:
@@ -1625,31 +1693,35 @@ def _import_annotations(request, rq_id, rq_func, pk, format_name):
     rq_job = queue.fetch_job(rq_id)
 
     if not rq_job:
-        serializer = AnnotationFileSerializer(data=request.data)
-        if serializer.is_valid(raise_exception=True):
-            anno_file = serializer.validated_data['annotation_file']
-            fd, filename = mkstemp(prefix='cvat_{}'.format(pk))
-            with open(filename, 'wb+') as f:
-                for chunk in anno_file.chunks():
-                    f.write(chunk)
-
-            av_scan_paths(filename)
-            rq_job = queue.enqueue_call(
-                func=rq_func,
-                args=(pk, filename, format_name),
-                job_id=rq_id
-            )
-            rq_job.meta['tmp_file'] = filename
-            rq_job.meta['tmp_file_descriptor'] = fd
-            rq_job.save_meta()
+        # If filename is specified we consider that file was uploaded via TUS, so it exists in filesystem
+        # Then we dont need to create temporary file
+        fd = None
+        if not filename:
+            serializer = AnnotationFileSerializer(data=request.data)
+            if serializer.is_valid(raise_exception=True):
+                anno_file = serializer.validated_data['annotation_file']
+                fd, filename = mkstemp(prefix='cvat_{}'.format(pk))
+                with open(filename, 'wb+') as f:
+                    for chunk in anno_file.chunks():
+                        f.write(chunk)
+
+        av_scan_paths(filename)
+        rq_job = queue.enqueue_call(
+            func=rq_func,
+            args=(pk, filename, format_name),
+            job_id=rq_id
+        )
+        rq_job.meta['tmp_file'] = filename
+        rq_job.meta['tmp_file_descriptor'] = fd
+        rq_job.save_meta()
     else:
         if rq_job.is_finished:
-            os.close(rq_job.meta['tmp_file_descriptor'])
+            if rq_job.meta['tmp_file_descriptor']: os.close(rq_job.meta['tmp_file_descriptor'])
             os.remove(rq_job.meta['tmp_file'])
             rq_job.delete()
             return Response(status=status.HTTP_201_CREATED)
         elif rq_job.is_failed:
-            os.close(rq_job.meta['tmp_file_descriptor'])
+            if rq_job.meta['tmp_file_descriptor']: os.close(rq_job.meta['tmp_file_descriptor'])
             os.remove(rq_job.meta['tmp_file'])
             exc_info = str(rq_job.exc_info)
             rq_job.delete()
diff --git a/cvat/apps/iam/filters.py b/cvat/apps/iam/filters.py
index 9b7d483efa54..98c789959e07 100644
--- a/cvat/apps/iam/filters.py
+++ b/cvat/apps/iam/filters.py
@@ -26,7 +26,7 @@ def filter_queryset(self, request, queryset, view):
         # objects inside an organization if the request in context of the
         # organization.
         visibility = request.iam_context['visibility']
-        if visibility and view.iam_organization_field:
+        if visibility and view.iam_organization_field and ('organization' in visibility):
             visibility[view.iam_organization_field] = visibility.pop('organization')
             return queryset.filter(**visibility).distinct()
         else:


Code review comments:
diff_hunk: @@ -304,6 +304,15 @@ def get_log_path(self):
     def get_task_artifacts_dirname(self):
         return os.path.join(self.get_task_dirname(), 'artifacts')
 
+    def get_tmp_dirname(self):
+        return os.path.join(self.get_task_dirname(), "tmp")
+
+    def get_tmp_file(self, filename):
+        tmp_dir = self.get_tmp_dirname()
+        if filename in os.listdir(tmp_dir) and os.path.isfile(os.path.join(tmp_dir, filename)):
path: cvat/apps/engine/models.py
text: @user1:
```suggestion
        if os.path.isfile(os.path.join(tmp_dir, filename)):
```

@author:
Done

diff_hunk: @@ -685,39 +685,75 @@ def jobs(self, request, pk):
 
         return Response(serializer.data)
 
-    def upload_finished(self, request):
+    # UploadMixin method
+    def can_upload(self, data_type):
+        if data_type == 'data':
+            db_model = self.get_object()
+            model_data = db_model.data
+            return model_data.size == 0
+        return True
+
+    # UploadMixin method
+    def get_upload_dir(self, data_type):
+        db_model = self.get_object()
+        if data_type == 'annotations':
+            return db_model.get_tmp_dirname()
+        elif data_type == 'data':
+            return db_model.data.get_upload_dirname()
+        return ""
+
+    # UploadMixin method
+    def upload_finished(self, request, data_type):
         db_task = self.get_object() # call check_object_permissions as well
path: cvat/apps/engine/views.py
text: @user1:
The method is called from `data` action. Thus self.get_object() was called. I will not recommend to call it again due to extra communication with OPA. Better to keep the object inside self. Something like self._object = self.get_object() and use it in tus methods.

@author:
Done

diff_hunk: @@ -685,39 +685,75 @@ def jobs(self, request, pk):
 
         return Response(serializer.data)
 
-    def upload_finished(self, request):
+    # UploadMixin method
+    def can_upload(self, data_type):
+        if data_type == 'data':
+            db_model = self.get_object()
+            model_data = db_model.data
+            return model_data.size == 0
+        return True
+
+    # UploadMixin method
+    def get_upload_dir(self, data_type):
+        db_model = self.get_object()
+        if data_type == 'annotations':
path: cvat/apps/engine/views.py
text: @user1:
In the context `self` is the view. Thus we have `self.action`. I don't think that we need to pass `data_type` at all.

@author:
Yes,`self.action` can help us on init/finish upload. But when we are uploading file with chunks, we use universal `append_tus_chunk` action where there is no direct information about `data_type`

@user1:
@author , probably it is because append_tus_chunk should be a method, not an action.

diff_hunk: @@ -685,39 +685,75 @@ def jobs(self, request, pk):
 
         return Response(serializer.data)
 
-    def upload_finished(self, request):
+    # UploadMixin method
+    def can_upload(self, data_type):
path: cvat/apps/engine/views.py
text: @user1:
We have a check `elif task_data.size != 0:` inside `data` action. Why do we need the extra check here? Also it will check permissions one more time due to call of self.get_object(). Is it really necessary?

@author:
It is some outdated check, removed it

diff_hunk: @@ -304,6 +304,15 @@ def get_log_path(self):
     def get_task_artifacts_dirname(self):
         return os.path.join(self.get_task_dirname(), 'artifacts')
 
+    def get_tmp_dirname(self):
+        return os.path.join(self.get_task_dirname(), "tmp")
+
+    def get_tmp_file(self, filename):
path: cvat/apps/engine/models.py
text: @user1:
Do we really need the method here?

@author:
Removed that method

diff_hunk: @@ -26,7 +26,7 @@ def filter_queryset(self, request, queryset, view):
         # objects inside an organization if the request in context of the
         # organization.
         visibility = request.iam_context['visibility']
-        if visibility and view.iam_organization_field:
+        if visibility and view.iam_organization_field and ('organization' in visibility):
path: cvat/apps/iam/filters.py
text: @user1:
@author , could you please provide a case, when the extra check is necessary? From code I see that visibility always has the field.

@author:
Yes, I've got such case with tus uploading of job annotations. When we use UploadMixin `self.get_object()` is called in different upload functions several times. At each call of this function we go to filter and on first call there is an 'organization' field in visibility, so everything works fine, but in code lines below:
```
if visibility and view.iam_organization_field and ('organization' in visibility)
  visibility[view.iam_organization_field] = visibility.pop('organization')
  return queryset.filter(**visibility).distinct()
```
'organization' field is removed and changed to 'segment__task__organization' (In JobView case). So in each next call of `self.get_object()` in UploadMixin there is no `organization` field and KeyError is raised without such extra check.