Unify error handling with the cloud storage

CHANGELOG.md

@@ -29,6 +29,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>)
 
 ### Deprecated
 - TDB

cvat/apps/engine/backup.py

@@ -23,6 +23,7 @@
 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.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 @@
  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 @@ def prepare_chunk_buff(self, db_data, quality, chunk_number):
  '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 @@
 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,16 +278,20 @@ def get_file_status(self, key):
  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,
  Key=file_name,
  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 @@ def initialize_content(self):
  '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 @@ def _head_file(self, key):
  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 @@ def get_file_status(self, key):
  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 @@ def initialize_content(self):
  '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,24 +555,24 @@ def initialize_content(self):
  )
  ]
 
+ @validate_file_status
+ @validate_bucket_status
  def download_fileobj(self, key):
  buf = BytesIO()
  blob = self.bucket.blob(key)
  self._storage_client.download_blob_to_file(blob, buf)
  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 @@ def create(self):
  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 @@ def mapping_with_new_values(self, 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,34 @@ 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 'PermissionDenied' in msg:
+ # msg like 'rest_framework.exceptions.PermissionDenied: ... \n'
+ parsed_msg = msg[44:-1]
+ elif 'NotFound' in msg:
+ # msg like: 'rest_framework.exceptions.NotFound: ... \n'
+ parsed_msg = msg[36:-1]
+ 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}'