Unify error handling with the cloud storage (#5389)

Added a general solution for checking bucket and file status instead of
checking in all places. Exception information has become more
user-friendly.

CHANGELOG.md

@@ -31,6 +31,7 @@ from online detectors & interactors) (<https://github.com/opencv/cvat/pull/4543>
 - Allowed trailing slashes in the SDK host address (<https://github.com/opencv/cvat/pull/5057>)
 - Adjusted initial camera position, enabled 'Reset zoom' option for 3D canvas (<https://github.com/opencv/cvat/pull/5395>)
 - Enabled authentication via email (<https://github.com/opencv/cvat/pull/5037>)
+- Unify error handling with the cloud storage (<https://github.com/opencv/cvat/pull/5389>)
 - In the SDK, functions taking paths as strings now also accept path-like objects
   (<https://github.com/opencv/cvat/pull/5435>)
 

cvat/apps/engine/backup.py

@@ -23,6 +23,7 @@ from rest_framework import serializers, status
 from rest_framework.parsers import JSONParser
 from rest_framework.renderers import JSONRenderer
 from rest_framework.response import Response
+from rest_framework.exceptions import ValidationError, PermissionDenied, NotFound
 from django_sendfile import sendfile
 from distutils.util import strtobool
 
@@ -39,9 +40,7 @@ from cvat.apps.engine.models import (
 from cvat.apps.engine.task import _create_thread
 from cvat.apps.dataset_manager.views import TASK_CACHE_TTL, PROJECT_CACHE_TTL, get_export_cache_dir, clear_export_cache, log_exception
 from cvat.apps.dataset_manager.bindings import CvatImportError
-from cvat.apps.engine.cloud_provider import (
-    db_storage_to_storage_instance, import_from_cloud_storage, export_to_cloud_storage
-)
+from cvat.apps.engine.cloud_provider import db_storage_to_storage_instance
 
 from cvat.apps.engine.location import StorageType, get_location_configuration
 
@@ -798,7 +797,12 @@ def export(db_instance, request):
                         db_storage = get_object_or_404(CloudStorageModel, pk=storage_id)
                         storage = db_storage_to_storage_instance(db_storage)
 
-                        export_to_cloud_storage(storage, file_path, filename)
+                        try:
+                            storage.upload_file(file_path, filename)
+                        except (ValidationError, PermissionDenied, NotFound) as ex:
+                            msg = str(ex) if not isinstance(ex, ValidationError) else \
+                                '\n'.join([str(d) for d in ex.detail])
+                            return Response(data=msg, status=ex.status_code)
                         return Response(status=status.HTTP_200_OK)
                     else:
                         raise NotImplementedError()
@@ -826,7 +830,7 @@ def export(db_instance, request):
 def _download_file_from_bucket(db_storage, filename, key):
     storage = db_storage_to_storage_instance(db_storage)
 
-    data = import_from_cloud_storage(storage, key)
+    data = storage.download_fileobj(key)
     with open(filename, 'wb+') as f:
         f.write(data.getbuffer())
 

cvat/apps/engine/cache.py

@@ -15,7 +15,7 @@ from cvat.apps.engine.media_extractors import (Mpeg4ChunkWriter,
     ImageDatasetManifestReader, VideoDatasetManifestReader)
 from cvat.apps.engine.models import DataChoice, StorageChoice
 from cvat.apps.engine.models import DimensionType
-from cvat.apps.engine.cloud_provider import get_cloud_storage_instance, Credentials, Status
+from cvat.apps.engine.cloud_provider import get_cloud_storage_instance, Credentials
 from cvat.apps.engine.utils import md5_hash
 class CacheInteraction:
     def __init__(self, dimension=DimensionType.DIM_2D):
@@ -82,36 +82,20 @@ class CacheInteraction:
                     'credentials': credentials,
                     'specific_attributes': db_cloud_storage.get_specific_attributes()
                 }
-                try:
-                    cloud_storage_instance = get_cloud_storage_instance(cloud_provider=db_cloud_storage.provider_type, **details)
-                    for item in reader:
-                        file_name = f"{item['name']}{item['extension']}"
-                        with NamedTemporaryFile(mode='w+b', prefix='cvat', suffix=file_name.replace(os.path.sep, '#'), delete=False) as temp_file:
-                            source_path = temp_file.name
-                            buf = cloud_storage_instance.download_fileobj(file_name)
-                            temp_file.write(buf.getvalue())
-                            temp_file.flush()
-                            checksum = item.get('checksum', None)
-                            if not checksum:
-                                slogger.cloud_storage[db_cloud_storage.id].warning('A manifest file does not contain checksum for image {}'.format(item.get('name')))
-                            if checksum and not md5_hash(source_path) == checksum:
-                                slogger.cloud_storage[db_cloud_storage.id].warning('Hash sums of files {} do not match'.format(file_name))
-                            images.append((source_path, source_path, None))
-                except Exception as ex:
-                    storage_status = cloud_storage_instance.get_status()
-                    if storage_status == Status.FORBIDDEN:
-                        msg = 'The resource {} is no longer available. Access forbidden.'.format(cloud_storage_instance.name)
-                    elif storage_status == Status.NOT_FOUND:
-                        msg = 'The resource {} not found. It may have been deleted.'.format(cloud_storage_instance.name)
-                    else:
-                        # check status of last file
-                        file_status = cloud_storage_instance.get_file_status(file_name)
-                        if file_status == Status.NOT_FOUND:
-                            raise Exception("'{}' not found on the cloud storage '{}'".format(file_name, cloud_storage_instance.name))
-                        elif file_status == Status.FORBIDDEN:
-                            raise Exception("Access to the file '{}' on the '{}' cloud storage is denied".format(file_name, cloud_storage_instance.name))
-                        msg = str(ex)
-                    raise Exception(msg)
+                cloud_storage_instance = get_cloud_storage_instance(cloud_provider=db_cloud_storage.provider_type, **details)
+                for item in reader:
+                    file_name = f"{item['name']}{item['extension']}"
+                    with NamedTemporaryFile(mode='w+b', prefix='cvat', suffix=file_name.replace(os.path.sep, '#'), delete=False) as temp_file:
+                        source_path = temp_file.name
+                        buf = cloud_storage_instance.download_fileobj(file_name)
+                        temp_file.write(buf.getvalue())
+                        temp_file.flush()
+                        checksum = item.get('checksum', None)
+                        if not checksum:
+                            slogger.cloud_storage[db_cloud_storage.id].warning('A manifest file does not contain checksum for image {}'.format(item.get('name')))
+                        if checksum and not md5_hash(source_path) == checksum:
+                            slogger.cloud_storage[db_cloud_storage.id].warning('Hash sums of files {} do not match'.format(file_name))
+                        images.append((source_path, source_path, None))
             else:
                 for item in reader:
                     source_path = os.path.join(upload_dir, f"{item['name']}{item['extension']}")

cvat/apps/engine/cloud_provider.py

@@ -10,7 +10,7 @@ import json
 from abc import ABC, abstractmethod, abstractproperty
 from enum import Enum
 from io import BytesIO
-from rest_framework import serializers
+from rest_framework.exceptions import PermissionDenied, NotFound, ValidationError
 
 from boto3.s3.transfer import TransferConfig
 from botocore.exceptions import ClientError
@@ -46,6 +46,45 @@ class Permissions(str, Enum):
     def all(cls):
         return {i.value for i in cls}
 
+
+def validate_bucket_status(func):
+    @functools.wraps(func)
+    def wrapper(self, *args, **kwargs):
+        try:
+            res = func(self, *args, **kwargs)
+        except Exception as ex:
+            # check that cloud storage exists
+            storage_status = self.get_status() if self is not None else None
+            if storage_status == Status.FORBIDDEN:
+                raise PermissionDenied('The resource {} is no longer available. Access forbidden.'.format(self.name))
+            elif storage_status == Status.NOT_FOUND:
+                raise NotFound('The resource {} not found. It may have been deleted.'.format(self.name))
+            elif storage_status == Status.AVAILABLE:
+                raise
+            raise ValidationError(str(ex))
+        return res
+    return wrapper
+
+def validate_file_status(func):
+    @functools.wraps(func)
+    def wrapper(self, *args, **kwargs):
+        try:
+            res = func(self, *args, **kwargs)
+        except Exception as ex:
+            storage_status = self.get_status() if self is not None else None
+            if storage_status == Status.AVAILABLE:
+                key = args[0]
+                file_status = self.get_file_status(key)
+                if file_status == Status.NOT_FOUND:
+                    raise NotFound("The file '{}' not found on the cloud storage '{}'".format(key, self.name))
+                elif file_status == Status.FORBIDDEN:
+                    raise PermissionDenied("Access to the file '{}' on the '{}' cloud storage is denied".format(key, self.name))
+                raise ValidationError(str(ex))
+            else:
+                raise
+        return res
+    return wrapper
+
 class _CloudStorage(ABC):
 
     def __init__(self):
@@ -239,9 +278,12 @@ class AWS_S3(_CloudStorage):
             else:
                 return Status.NOT_FOUND
 
+    @validate_file_status
+    @validate_bucket_status
     def get_file_last_modified(self, key):
         return self._head_file(key).get('LastModified')
 
+    @validate_bucket_status
     def upload_fileobj(self, file_obj, file_name):
         self._bucket.upload_fileobj(
             Fileobj=file_obj,
@@ -249,6 +291,7 @@ class AWS_S3(_CloudStorage):
             Config=TransferConfig(max_io_queue=self.transfer_config['max_io_queue'])
         )
 
+    @validate_bucket_status
     def upload_file(self, file_path, file_name=None):
         if not file_name:
             file_name = os.path.basename(file_path)
@@ -269,6 +312,8 @@ class AWS_S3(_CloudStorage):
             'name': item.key,
         } for item in files]
 
+    @validate_file_status
+    @validate_bucket_status
     def download_fileobj(self, key):
         buf = BytesIO()
         self.bucket.download_fileobj(
@@ -378,6 +423,8 @@ class AzureBlobContainer(_CloudStorage):
         blob_client = self.container.get_blob_client(key)
         return blob_client.get_blob_properties()
 
+    @validate_file_status
+    @validate_bucket_status
     def get_file_last_modified(self, key):
         return self._head_file(key).last_modified
 
@@ -401,18 +448,15 @@ class AzureBlobContainer(_CloudStorage):
             else:
                 return Status.NOT_FOUND
 
+    @validate_bucket_status
     def upload_fileobj(self, file_obj, file_name):
         self._container_client.upload_blob(name=file_name, data=file_obj)
 
     def upload_file(self, file_path, file_name=None):
         if not file_name:
             file_name = os.path.basename(file_path)
-        try:
-            with open(file_path, 'r') as f:
-                self.upload_fileobj(f, file_name)
-        except Exception as ex:
-            slogger.glob.error(str(ex))
-            raise
+        with open(file_path, 'r') as f:
+            self.upload_fileobj(f, file_name)
 
     # TODO:
     # def multipart_upload(self, file_obj):
@@ -424,6 +468,8 @@ class AzureBlobContainer(_CloudStorage):
             'name': item.name
         } for item in files]
 
+    @validate_file_status
+    @validate_bucket_status
     def download_fileobj(self, key):
         buf = BytesIO()
         storage_stream_downloader = self._container_client.download_blob(
@@ -509,6 +555,8 @@ class GoogleCloudStorage(_CloudStorage):
             )
         ]
 
+    @validate_file_status
+    @validate_bucket_status
     def download_fileobj(self, key):
         buf = BytesIO()
         blob = self.bucket.blob(key)
@@ -516,17 +564,15 @@ class GoogleCloudStorage(_CloudStorage):
         buf.seek(0)
         return buf
 
+    @validate_bucket_status
     def upload_fileobj(self, file_obj, file_name):
         self.bucket.blob(file_name).upload_from_file(file_obj)
 
+    @validate_bucket_status
     def upload_file(self, file_path, file_name=None):
         if not file_name:
             file_name = os.path.basename(file_path)
-        try:
-            self.bucket.blob(file_name).upload_from_filename(file_path)
-        except Exception as ex:
-            slogger.glob.info(str(ex))
-            raise
+        self.bucket.blob(file_name).upload_from_filename(file_path)
 
     def create(self):
         try:
@@ -545,6 +591,8 @@ class GoogleCloudStorage(_CloudStorage):
             slogger.glob.info(msg)
             raise Exception(msg)
 
+    @validate_file_status
+    @validate_bucket_status
     def get_file_last_modified(self, key):
         blob = self.bucket.blob(key)
         blob.reload()
@@ -616,26 +664,6 @@ class Credentials:
     def values(self):
         return [self.key, self.secret_key, self.session_token, self.account_name, self.key_file_path]
 
-
-def validate_bucket_status(func):
-    @functools.wraps(func)
-    def wrapper(storage, *args, **kwargs):
-        try:
-            res = func(storage, *args, **kwargs)
-        except Exception as ex:
-            # check that cloud storage exists
-            storage_status = storage.get_status() if storage is not None else None
-            if storage_status == Status.FORBIDDEN:
-                msg = 'The resource {} is no longer available. Access forbidden.'.format(storage.name)
-            elif storage_status == Status.NOT_FOUND:
-                msg = 'The resource {} not found. It may have been deleted.'.format(storage.name)
-            else:
-                msg = str(ex)
-            raise serializers.ValidationError(msg)
-        return res
-    return wrapper
-
-
 def db_storage_to_storage_instance(db_storage):
     credentials = Credentials()
     credentials.convert_from_db({
@@ -648,11 +676,3 @@ def db_storage_to_storage_instance(db_storage):
         'specific_attributes': db_storage.get_specific_attributes()
     }
     return get_cloud_storage_instance(cloud_provider=db_storage.provider_type, **details)
-
-@validate_bucket_status
-def import_from_cloud_storage(storage, file_name):
-    return storage.download_fileobj(file_name)
-
-@validate_bucket_status
-def export_to_cloud_storage(storage, file_path, file_name):
-    storage.upload_file(file_path, file_name)

cvat/apps/engine/utils.py

@@ -110,17 +110,30 @@ def parse_specific_attributes(specific_attributes):
     } if parsed_specific_attributes else dict()
 
 
+def parse_exception_message(msg):
+    parsed_msg = msg
+    try:
+        if 'ErrorDetail' in msg:
+            # msg like: 'rest_framework.exceptions.ValidationError:
+            # [ErrorDetail(string="...", code=\'invalid\')]\n'
+            parsed_msg = msg.split('string=')[1].split(', code=')[0].strip("\"")
+        elif msg.startswith('rest_framework.exceptions.'):
+            parsed_msg = msg.split(':')[1].strip()
+    except Exception: # nosec
+        pass
+    return parsed_msg
+
 def process_failed_job(rq_job):
     if rq_job.meta['tmp_file_descriptor']:
         os.close(rq_job.meta['tmp_file_descriptor'])
     if os.path.exists(rq_job.meta['tmp_file']):
         os.remove(rq_job.meta['tmp_file'])
-    exc_info = str(rq_job.exc_info) or str(rq_job.dependency.exc_info)
+    exc_info = str(rq_job.exc_info or rq_job.dependency.exc_info)
     if rq_job.dependency:
         rq_job.dependency.delete()
     rq_job.delete()
 
-    return exc_info
+    return parse_exception_message(exc_info)
 
 def configure_dependent_job(queue, rq_id, rq_func, db_storage, filename, key):
     rq_job_id_download_file = rq_id + f'?action=download_{filename}'

cvat/apps/engine/views.py

@@ -3,7 +3,6 @@
 #
 # SPDX-License-Identifier: MIT
 
-import errno
 import io
 import os
 import os.path as osp
@@ -37,19 +36,15 @@ from drf_spectacular.plumbing import build_array_type, build_basic_type
 
 from rest_framework import mixins, serializers, status, viewsets
 from rest_framework.decorators import action
-from rest_framework.exceptions import APIException, NotFound, ValidationError
+from rest_framework.exceptions import APIException, NotFound, ValidationError, PermissionDenied
 from rest_framework.permissions import SAFE_METHODS
 from rest_framework.renderers import JSONRenderer
 from rest_framework.response import Response
-from rest_framework.exceptions import PermissionDenied
 from django_sendfile import sendfile
 
 import cvat.apps.dataset_manager as dm
 import cvat.apps.dataset_manager.views  # pylint: disable=unused-import
-from cvat.apps.engine.cloud_provider import (
-    db_storage_to_storage_instance, import_from_cloud_storage, export_to_cloud_storage,
-    Status as CloudStorageStatus
-)
+from cvat.apps.engine.cloud_provider import db_storage_to_storage_instance
 from cvat.apps.dataset_manager.bindings import CvatImportError
 from cvat.apps.dataset_manager.serializers import DatasetFormatsSerializer
 from cvat.apps.engine.frame_provider import FrameProvider
@@ -73,7 +68,9 @@ from cvat.apps.engine.serializers import (
     ProjectFileSerializer, TaskFileSerializer)
 
 from utils.dataset_manifest import ImageManifestManager
-from cvat.apps.engine.utils import av_scan_paths, process_failed_job, configure_dependent_job
+from cvat.apps.engine.utils import (
+    av_scan_paths, process_failed_job, configure_dependent_job, parse_exception_message
+)
 from cvat.apps.engine import backup
 from cvat.apps.engine.mixins import PartialUpdateModelMixin, UploadMixin, AnnotationMixin, SerializeMixin, DestroyModelMixin, CreateModelMixin
 from cvat.apps.engine.location import get_location_configuration, StorageType
@@ -674,53 +671,58 @@ class DataChunkGetter:
 
         frame_provider = FrameProvider(db_data, self.dimension)
 
-        if self.type == 'chunk':
-            start_chunk = frame_provider.get_chunk_number(start)
-            stop_chunk = frame_provider.get_chunk_number(stop)
-            # pylint: disable=superfluous-parens
-            if not (start_chunk <= self.number <= stop_chunk):
-                raise ValidationError('The chunk number should be in ' +
-                    f'[{start_chunk}, {stop_chunk}] range')
-
-            # TODO: av.FFmpegError processing
-            if settings.USE_CACHE and db_data.storage_method == StorageMethodChoice.CACHE:
-                buff, mime_type = frame_provider.get_chunk(self.number, self.quality)
-                return HttpResponse(buff.getvalue(), content_type=mime_type)
-
-            # Follow symbol links if the chunk is a link on a real image otherwise
-            # mimetype detection inside sendfile will work incorrectly.
-            path = os.path.realpath(frame_provider.get_chunk(self.number, self.quality))
-            return sendfile(request, path)
-
-        elif self.type == 'frame':
-            if not (start <= self.number <= stop):
-                raise ValidationError('The frame number should be in ' +
-                    f'[{start}, {stop}] range')
-
-            buf, mime = frame_provider.get_frame(self.number, self.quality)
-            return HttpResponse(buf.getvalue(), content_type=mime)
-
-        elif self.type == 'preview':
-            return sendfile(request, db_object.get_preview_path())
-
-        elif self.type == 'context_image':
-            if not (start <= self.number <= stop):
-                raise ValidationError('The frame number should be in ' +
-                    f'[{start}, {stop}] range')
-
-            image = Image.objects.get(data_id=db_data.id, frame=self.number)
-            for i in image.related_files.all():
-                path = os.path.realpath(str(i.path))
-                image = cv2.imread(path)
-                success, result = cv2.imencode('.JPEG', image)
-                if not success:
-                    raise Exception('Failed to encode image to ".jpeg" format')
-                return HttpResponse(io.BytesIO(result.tobytes()), content_type='image/jpeg')
-            return Response(data='No context image related to the frame',
-                status=status.HTTP_404_NOT_FOUND)
-        else:
-            return Response(data='unknown data type {}.'.format(self.type),
-                status=status.HTTP_400_BAD_REQUEST)
+        try:
+            if self.type == 'chunk':
+                start_chunk = frame_provider.get_chunk_number(start)
+                stop_chunk = frame_provider.get_chunk_number(stop)
+                # pylint: disable=superfluous-parens
+                if not (start_chunk <= self.number <= stop_chunk):
+                    raise ValidationError('The chunk number should be in ' +
+                        f'[{start_chunk}, {stop_chunk}] range')
+
+                # TODO: av.FFmpegError processing
+                if settings.USE_CACHE and db_data.storage_method == StorageMethodChoice.CACHE:
+                    buff, mime_type = frame_provider.get_chunk(self.number, self.quality)
+                    return HttpResponse(buff.getvalue(), content_type=mime_type)
+
+                # Follow symbol links if the chunk is a link on a real image otherwise
+                # mimetype detection inside sendfile will work incorrectly.
+                path = os.path.realpath(frame_provider.get_chunk(self.number, self.quality))
+                return sendfile(request, path)
+
+            elif self.type == 'frame':
+                if not (start <= self.number <= stop):
+                    raise ValidationError('The frame number should be in ' +
+                        f'[{start}, {stop}] range')
+
+                buf, mime = frame_provider.get_frame(self.number, self.quality)
+                return HttpResponse(buf.getvalue(), content_type=mime)
+
+            elif self.type == 'preview':
+                return sendfile(request, db_object.get_preview_path())
+
+            elif self.type == 'context_image':
+                if not (start <= self.number <= stop):
+                    raise ValidationError('The frame number should be in ' +
+                        f'[{start}, {stop}] range')
+
+                image = Image.objects.get(data_id=db_data.id, frame=self.number)
+                for i in image.related_files.all():
+                    path = os.path.realpath(str(i.path))
+                    image = cv2.imread(path)
+                    success, result = cv2.imencode('.JPEG', image)
+                    if not success:
+                        raise Exception('Failed to encode image to ".jpeg" format')
+                    return HttpResponse(io.BytesIO(result.tobytes()), content_type='image/jpeg')
+                return Response(data='No context image related to the frame',
+                    status=status.HTTP_404_NOT_FOUND)
+            else:
+                return Response(data='unknown data type {}.'.format(self.type),
+                    status=status.HTTP_400_BAD_REQUEST)
+        except (ValidationError, PermissionDenied, NotFound) as ex:
+            msg = str(ex) if not isinstance(ex, ValidationError) else \
+                '\n'.join([str(d) for d in ex.detail])
+            return Response(data=msg, status=ex.status_code)
 
 @extend_schema(tags=['tasks'])
 @extend_schema_view(
@@ -1225,7 +1227,7 @@ class TaskViewSet(viewsets.GenericViewSet, mixins.ListModelMixin,
             # It's not really clear how it is possible, but it can
             # lead to an error in serializing the response
             # https://github.com/opencv/cvat/issues/5215
-            response = { "state": "Failed", "message": job.exc_info or "Unknown error" }
+            response = { "state": "Failed", "message": parse_exception_message(job.exc_info or "Unknown error") }
         else:
             response = { "state": "Started" }
             if job.meta.get('status'):
@@ -2068,16 +2070,9 @@ class CloudStorageViewSet(viewsets.GenericViewSet, mixins.ListModelMixin,
             db_storage = self.get_object()
             storage = db_storage_to_storage_instance(db_storage)
             if not db_storage.manifests.count():
-                raise Exception('There is no manifest file')
+                raise ValidationError('There is no manifest file')
             manifest_path = request.query_params.get('manifest_path', db_storage.manifests.first().filename)
             manifest_prefix = os.path.dirname(manifest_path)
-            file_status = storage.get_file_status(manifest_path)
-            if file_status == CloudStorageStatus.NOT_FOUND:
-                raise FileNotFoundError(errno.ENOENT,
-                    "Not found on the cloud storage {}".format(db_storage.display_name), manifest_path)
-            elif file_status == CloudStorageStatus.FORBIDDEN:
-                raise PermissionError(errno.EACCES,
-                    "Access to the file on the '{}' cloud storage is denied".format(db_storage.display_name), manifest_path)
 
             full_manifest_path = os.path.join(db_storage.get_storage_dirname(), manifest_path)
             if not os.path.exists(full_manifest_path) or \
@@ -2093,20 +2088,15 @@ class CloudStorageViewSet(viewsets.GenericViewSet, mixins.ListModelMixin,
             message = f"Storage {pk} does not exist"
             slogger.glob.error(message)
             return HttpResponseNotFound(message)
-        except FileNotFoundError as ex:
-            msg = f"{ex.strerror} {ex.filename}"
+        except (ValidationError, PermissionDenied, NotFound) as ex:
+            msg = str(ex) if not isinstance(ex, ValidationError) else \
+                '\n'.join([str(d) for d in ex.detail])
             slogger.cloud_storage[pk].info(msg)
-            return Response(data=msg, status=status.HTTP_404_NOT_FOUND)
+            return Response(data=msg, status=ex.status_code)
         except Exception as ex:
-            # check that cloud storage was not deleted
-            storage_status = storage.get_status() if storage else None
-            if storage_status == CloudStorageStatus.FORBIDDEN:
-                msg = 'The resource {} is no longer available. Access forbidden.'.format(storage.name)
-            elif storage_status == CloudStorageStatus.NOT_FOUND:
-                msg = 'The resource {} not found. It may have been deleted.'.format(storage.name)
-            else:
-                msg = str(ex)
-            return HttpResponseBadRequest(msg)
+            slogger.glob.error(str(ex))
+            return Response("An internal error has occurred",
+                status=status.HTTP_500_INTERNAL_SERVER_ERROR)
 
     @extend_schema(summary='Method returns a preview image from a cloud storage',
         responses={
@@ -2120,7 +2110,7 @@ class CloudStorageViewSet(viewsets.GenericViewSet, mixins.ListModelMixin,
             if not os.path.exists(db_storage.get_preview_path()):
                 storage = db_storage_to_storage_instance(db_storage)
                 if not db_storage.manifests.count():
-                    raise Exception('Cannot get the cloud storage preview. There is no manifest file')
+                    raise ValidationError('Cannot get the cloud storage preview. There is no manifest file')
                 preview_path = None
                 for manifest_model in db_storage.manifests.all():
                     manifest_prefix = os.path.dirname(manifest_model.filename)
@@ -2145,13 +2135,6 @@ class CloudStorageViewSet(viewsets.GenericViewSet, mixins.ListModelMixin,
                     slogger.cloud_storage[pk].info(msg)
                     return HttpResponseBadRequest(msg)
 
-                file_status = storage.get_file_status(preview_path)
-                if file_status == CloudStorageStatus.NOT_FOUND:
-                    raise FileNotFoundError(errno.ENOENT,
-                        "Not found on the cloud storage {}".format(db_storage.display_name), preview_path)
-                elif file_status == CloudStorageStatus.FORBIDDEN:
-                    raise PermissionError(errno.EACCES,
-                        "Access to the file on the '{}' cloud storage is denied".format(db_storage.display_name), preview_path)
                 with NamedTemporaryFile() as temp_image:
                     storage.download_file(preview_path, temp_image.name)
                     reader = ImageListReader([temp_image.name])
@@ -2163,18 +2146,15 @@ class CloudStorageViewSet(viewsets.GenericViewSet, mixins.ListModelMixin,
             message = f"Storage {pk} does not exist"
             slogger.glob.error(message)
             return HttpResponseNotFound(message)
-        except PermissionDenied:
-            raise
+        except (ValidationError, PermissionDenied, NotFound) as ex:
+            msg = str(ex) if not isinstance(ex, ValidationError) else \
+                '\n'.join([str(d) for d in ex.detail])
+            slogger.cloud_storage[pk].info(msg)
+            return Response(data=msg, status=ex.status_code)
         except Exception as ex:
-            # check that cloud storage was not deleted
-            storage_status = storage.get_status() if storage else None
-            if storage_status == CloudStorageStatus.FORBIDDEN:
-                msg = 'The resource {} is no longer available. Access forbidden.'.format(storage.name)
-            elif storage_status == CloudStorageStatus.NOT_FOUND:
-                msg = 'The resource {} not found. It may have been deleted.'.format(storage.name)
-            else:
-                msg = str(ex)
-            return HttpResponseBadRequest(msg)
+            slogger.glob.error(str(ex))
+            return Response("An internal error has occurred",
+                status=status.HTTP_500_INTERNAL_SERVER_ERROR)
 
     @extend_schema(summary='Method returns a cloud storage status',
         responses={
@@ -2186,7 +2166,7 @@ class CloudStorageViewSet(viewsets.GenericViewSet, mixins.ListModelMixin,
             db_storage = self.get_object()
             storage = db_storage_to_storage_instance(db_storage)
             storage_status = storage.get_status()
-            return HttpResponse(storage_status)
+            return Response(storage_status)
         except CloudStorageModel.DoesNotExist:
             message = f"Storage {pk} does not exist"
             slogger.glob.error(message)
@@ -2229,7 +2209,7 @@ def rq_handler(job, exc_type, exc_value, tb):
 def _download_file_from_bucket(db_storage, filename, key):
     storage = db_storage_to_storage_instance(db_storage)
 
-    data = import_from_cloud_storage(storage, key)
+    data = storage.download_fileobj(key)
     with open(filename, 'wb+') as f:
         f.write(data.getbuffer())
 
@@ -2367,7 +2347,12 @@ def _export_annotations(db_instance, rq_id, request, format_name, action, callba
                         db_storage = get_object_or_404(CloudStorageModel, pk=storage_id)
                         storage = db_storage_to_storage_instance(db_storage)
 
-                        export_to_cloud_storage(storage, file_path, filename)
+                        try:
+                            storage.upload_file(file_path, filename)
+                        except (ValidationError, PermissionDenied, NotFound) as ex:
+                            msg = str(ex) if not isinstance(ex, ValidationError) else \
+                                '\n'.join([str(d) for d in ex.detail])
+                            return Response(data=msg, status=ex.status_code)
                         return Response(status=status.HTTP_200_OK)
                     else:
                         raise NotImplementedError()

tests/python/rest_api/test_tasks.py

@@ -432,6 +432,21 @@ class TestPostTaskData:
 
         return task.id
 
+    def _test_cannot_create_task(self, username, spec, data, **kwargs):
+        with make_api_client(username) as api_client:
+            (task, response) = api_client.tasks_api.create(spec, **kwargs)
+            assert response.status == HTTPStatus.CREATED
+
+            (_, response) = api_client.tasks_api.create_data(
+                task.id, data_request=deepcopy(data), _content_type="application/json", **kwargs
+            )
+            assert response.status == HTTPStatus.ACCEPTED
+
+            status = self._wait_until_task_is_created(api_client.tasks_api, task.id)
+            assert status.state.value == "Failed"
+
+        return status
+
     def test_can_create_task_with_defined_start_and_stop_frames(self):
         task_spec = {
             "name": f"test {self._USERNAME} to create a task with defined start and stop frames",
@@ -651,7 +666,6 @@ class TestPostTaskData:
         data_spec = {
             "image_quality": 75,
             "use_cache": True,
-            "storage": "cloud_storage",
             "cloud_storage_id": cloud_storage_id,
             "server_files": cloud_storage_content,
         }
@@ -659,3 +673,28 @@ class TestPostTaskData:
         self._test_create_task(
             self._USERNAME, task_spec, data_spec, content_type="application/json", org=org
         )
+
+    @pytest.mark.parametrize(
+        "cloud_storage_id, manifest, org",
+        [(1, "manifest.jsonl", "")],  # public bucket
+    )
+    def test_cannot_create_task_with_mythical_cloud_storage_data(
+        self, cloud_storage_id, manifest, org
+    ):
+        mythical_file = "mythical.jpg"
+        cloud_storage_content = [mythical_file, manifest]
+
+        task_spec = {
+            "name": f"Task with mythical file from cloud storage {cloud_storage_id}",
+            "labels": [{"name": "car"}],
+        }
+
+        data_spec = {
+            "image_quality": 75,
+            "use_cache": True,
+            "cloud_storage_id": cloud_storage_id,
+            "server_files": cloud_storage_content,
+        }
+
+        status = self._test_cannot_create_task(self._USERNAME, task_spec, data_spec, org=org)
+        assert mythical_file in status.message