Cancel or delete stale pending and failed exports

onadata/apps/viewer/tasks.py

@@ -1,6 +1,8 @@
 import sys
+from datetime import datetime, timedelta
 
 from celery import task
+from celery.task.schedules import crontab
 from django.conf import settings
 from django.shortcuts import get_object_or_404
 from requests import ConnectionError
@@ -349,3 +351,39 @@ def delete_export(export_id):
         export.delete()
         return True
     return False
+
+
+@task.periodic_task(
+    run_every=(crontab(hour='*/7')),
+    ignore_result=True
+)
+def check_pending_exports():
+    """
+    Exports that have not completed within a set time should be marked as
+    failed
+    """
+    h = settings.EXPORT_TASK_LIFESPAN
+    time_threshold = datetime.now() - timedelta(hours=h)
+    exports = Export.objects.filter(internal_status=Export.PENDING,
+                                    created_on__lt=time_threshold)
+    for export in exports:
+        export.internal_status = Export.FAILED
+        export.save()
+    return True
+
+
+@task.periodic_task(
+    run_every=(crontab(hour='*/7')),
+    ignore_result=True
+)
+def delete_old_failed_exports():
+    """
+    Delete old failed exports
+    """
+    h = settings.EXPORT_TASK_LIFESPAN
+    time_threshold = datetime.now() - timedelta(hours=h)
+    exports = Export.objects.filter(internal_status=Export.FAILED,
+                                    created_on__lt=time_threshold)
+    for export in exports:
+        delete_export.delay(export.id)
+    return True

onadata/apps/viewer/tests/test_tasks.py

@@ -4,6 +4,8 @@
 from onadata.apps.main.tests.test_base import TestBase
 from onadata.apps.viewer.models.export import Export
 from onadata.apps.viewer.tasks import create_async_export
+from onadata.apps.viewer.tasks import check_pending_exports
+from onadata.apps.viewer.tasks import delete_old_failed_exports
 
 
 class TestExportTasks(TestBase):
@@ -38,3 +40,11 @@ def test_create_async(self):
             self.assertTrue(export.id)
             self.assertIn("username", options)
             self.assertEquals(options.get("id_string"), self.xform.id_string)
+
+    def test_check_pending_exports(self):
+        result = check_pending_exports.delay()
+        self.assertTrue(result)
+
+    def test_delete_old_failed_exports(self):
+        result = delete_old_failed_exports.delay()
+        self.assertTrue(result)

onadata/settings/common.py

@@ -435,6 +435,7 @@ def configure_logging(logger, **kwargs):
 
 # number of records on export or CSV import before a progress update
 EXPORT_TASK_PROGRESS_UPDATE_BATCH = 1000
+EXPORT_TASK_LIFESPAN = 6  # six hours
 
 # default content length for submission requests
 DEFAULT_CONTENT_LENGTH = 10000000