Finish document upload task refactor. Increase failure tolerance to database Operational Errors.

mayan/apps/sources/tasks.py

@@ -1,14 +1,17 @@
 import logging
 
 from django.contrib.auth.models import User
-from django.db import OperationalError
+from django.core.files import File
+from django.db import OperationalError, transaction
 from django.utils.translation import ugettext_lazy as _
 
 from mayan.celery import app
 
+from common.compressed_files import CompressedFile, NotACompressedFile
 from common.models import SharedUploadedFile
 from converter.models import Transformation
-from documents.models import DocumentType
+from documents.models import Document, DocumentType
+from documents.settings import setting_language
 from metadata.api import save_metadata_list
 
 from .literals import DEFAULT_SOURCE_TASK_RETRY_DELAY
@@ -31,23 +34,36 @@ def task_check_interval_source(source_id):
 
 
 @app.task(bind=True, default_retry_delay=DEFAULT_SOURCE_TASK_RETRY_DELAY, ignore_result=True)
-def task_post_document_version_upload(self, source_id, document_version_id):
+def task_upload_document(self, source_id, document_type_id, shared_uploaded_file_id, description=None, label=None, language=None, metadata_dict_list=None, user_id=None):
     try:
+        document_type = DocumentType.objects.get(pk=document_type_id)
         source = Source.objects.get_subclass(pk=source_id)
-        document_version = DocumentVersion.objects.get(pk=document_version_id)
+        shared_upload = SharedUploadedFile.objects.get(pk=shared_uploaded_file_id)
+
+        if user_id:
+            user = User.objects.get(pk=user_id)
+        else:
+            user = None
+
+        with shared_upload.open() as file_object:
+            source.upload_document(file_object=file_object, document_type=document_type, description=description, label=label, language=language, metadata_dict_list=metadata_dict_list, user=user)
 
-        Transformation.objects.copy(source=Source.objects.get_subclass(pk=source_id), targets=document_version.pages.all())
     except OperationalError as exception:
-        logger.warning('Operational error during post source document upload processing: %s. Retrying.', exception)
+        logger.warning('Operational exception while trying to create new document "%s" from source id %d; %s. Retying.', label or shared_upload.filename, source_id, exception)
         raise self.retry(exc=exception)
+    else:
+        try:
+            shared_upload.delete()
+        except OperationalError as exception:
+            logger.warning('Operational error during attempt to delete shared upload file: %s; %s. Retrying.', shared_upload, exception)
 
 
 @app.task(bind=True, default_retry_delay=DEFAULT_SOURCE_TASK_RETRY_DELAY, ignore_result=True)
-def task_upload_document(self, document_type_id, shared_uploaded_file_id, label=None, language=None, user_id=None, description=None, metadata_dict_list=None):
+def task_source_handle_upload(self, document_type_id, shared_uploaded_file_id, source_id, description=None, expand=False, label=None, language=None, metadata_dict_list=None, skip_list=None, user_id=None):
     try:
-        shared_uploaded_file = SharedUploadedFile.objects.get(pk=shared_uploaded_file_id)
-        source = Source.objects.get_subclass(pk=source_id)
         document_type = DocumentType.objects.get(pk=document_type_id)
+        source = Source.objects.get_subclass(pk=source_id)
+        shared_upload = SharedUploadedFile.objects.get(pk=shared_uploaded_file_id)
 
         if user_id:
             user = User.objects.get(pk=user_id)
@@ -55,62 +71,60 @@ def task_upload_document(self, document_type_id, shared_uploaded_file_id, label=
             user = None
 
         if not label:
-            label = shared_uploaded_file.filename
+            label = shared_upload.filename
 
-        with transaction.atomic():
-            document = DocumentVersion.objects.create(document_type=document_type)
-
-            document_version = document.new_document(
-                file_object=file_object, label=label, description=description,
-                language=language, _user=user
-            )
-
-            if metadata_dict_list:
-                save_metadata_list(metadata_dict_list, document, create=True)
-
-            task_post_source_document_version_upload.delay(source_id=source_id, document_version_id=document_version.pk, metadata_dict_list=metadata_dict_list)
-    except OperationalError as exception:
-        logger.warning('Operational error during attempt to handle source upload: %s. Retrying.', exception)
-        raise self.retry(exc=exception)
-
-    try:
-        shared_uploaded_file.delete()
-    except OperationalError as exception:
-        logger.warning('Operational error during attempt to delete shared upload file: %s; %s. Retrying.', shared_uploaded_file, exception)
-
-
-@app.task(bind=True, default_retry_delay=DEFAULT_SOURCE_TASK_RETRY_DELAY, ignore_result=True)
-def task_source_handle_upload(self, label, document_type_id, shared_uploaded_file_id, source_id, description=None, expand=False, language=None, metadata_dict_list=None, user_id=None):
-    try:
-        shared_uploaded_file = SharedUploadedFile.objects.get(pk=shared_uploaded_file_id)
-        source = Source.objects.get_subclass(pk=source_id)
-        document_type = DocumentType.objects.get(pk=document_type_id)
-
-        if user_id:
-            user = User.objects.get(pk=user_id)
-        else:
-            user = None
-
-        if not label:
-            label = shared_uploaded_file.filename
     except OperationalError as exception:
         logger.warning('Operational error during attempt to load data to handle source upload: %s. Retrying.', exception)
         raise self.retry(exc=exception)
 
-    with shared_uploaded_file.open() as file_object:
-        source.handle_upload(description=description, document_type=document_type, expand=expand, file_object=file_object, label=label, language=language, metadata_dict_list=metadata_dict_list, user=user)
+    kwargs = {
+        'description': description, 'document_type_id': document_type.pk,
+        'label': label, 'language': language,
+        'metadata_dict_list': metadata_dict_list,
+        'source_id': source_id, 'user_id': user_id
+    }
 
-    try:
-        shared_uploaded_file.delete()
-    except OperationalError as exception:
-        logger.warning('Operational error during attempt to delete shared upload file: %s; %s. Retrying.', shared_uploaded_file, exception)
+    if not skip_list:
+        skip_list = []
 
-    # TODO: Report/record how was file uploaded
-    #    if result['is_compressed'] is None:
-    #        messages.success(request, _('File uploaded successfully.'))
+    with shared_upload.open() as file_object:
+        if expand:
+            try:
+                compressed_file = CompressedFile(file_object)
+                for compressed_file_child in compressed_file.children():
+                    # TODO: find way to uniquely identify child files
+                    # Use filename in the mean time.
+                    if unicode(compressed_file_child) not in skip_list:
+                        kwargs.update({'label': unicode(compressed_file_child)})
 
-    #    if result['is_compressed'] is True:
-    #        messages.success(request, _('File uncompressed successfully and uploaded as individual files.'))
+                        try:
+                            child_shared_uploaded_file = SharedUploadedFile.objects.create(file=File(compressed_file_child))
+                        except OperationalError as exception:
+                            logger.warning('Operational error while preparing to upload child document: %s. Rescheduling.', exception)
 
-    #    if result['is_compressed'] is False:
-    #        messages.warning(request, _('File was not a compressed file, uploaded as it was.'))
+                            task_source_handle_upload.delay(
+                                document_type_id=document_type_id,
+                                shared_uploaded_file_id=shared_uploaded_file_id,
+                                source_id=source_id, description=description,
+                                expand=expand, label=label,
+                                language=language,
+                                metadata_dict_list=metadata_dict_list,
+                                skip_list=skip_list, user_id=user_id
+                            )
+                            return
+                        else:
+                            skip_list.append(unicode(compressed_file_child))
+                            task_upload_document.delay(shared_uploaded_file_id=child_shared_uploaded_file.pk, **kwargs)
+                        finally:
+                            compressed_file_child.close()
+
+                    compressed_file_child.close()
+                try:
+                    shared_upload.delete()
+                except OperationalError as exception:
+                    logger.warning('Operational error during attempt to delete shared upload file: %s; %s. Retrying.', shared_uploaded_file, exception)
+            except NotACompressedFile:
+                logging.debug('Exception: NotACompressedFile')
+                task_upload_document.delay(shared_uploaded_file_id=shared_upload.pk, **kwargs)
+        else:
+            task_upload_document.delay(shared_uploaded_file_id=shared_upload.pk, **kwargs)