Make fetch job deletion a Celery task. (#40)

Move fetch job deletion logic into a Celery task to avoid timeouts.

AIPscan/Aggregator/tasks.py

@@ -1,6 +1,7 @@
 # -*- coding: utf-8 -*-
 import json
 import os
+import shutil
 
 import requests
 from celery.utils.log import get_task_logger
@@ -20,7 +21,7 @@
 )
 from AIPscan.extensions import celery
 from AIPscan.helpers import file_sha256_hash
-from AIPscan.models import AIP, get_mets_tasks  # Custom celery Models.
+from AIPscan.models import AIP, FetchJob, get_mets_tasks  # Custom celery Models.
 
 logger = get_task_logger(__name__)
 
@@ -373,3 +374,12 @@ def get_mets(
         os.remove(download_file)
     except OSError as err:
         tasklogger.warning("Unable to delete METS file: {}".format(err))
+
+
+@celery.task()
+def delete_fetch_job(fetch_job_id):
+    fetch_job = FetchJob.query.get(fetch_job_id)
+    if os.path.exists(fetch_job.download_directory):
+        shutil.rmtree(fetch_job.download_directory)
+    db.session.delete(fetch_job)
+    db.session.commit()

AIPscan/Aggregator/tests/test_tasks.py

@@ -11,6 +11,7 @@
 from AIPscan.Aggregator.tasks import (
     TaskError,
     delete_aip,
+    delete_fetch_job,
     get_mets,
     make_request,
     parse_packages_and_load_mets,
@@ -23,7 +24,7 @@
     VALID_JSON,
     MockResponse,
 )
-from AIPscan.models import AIP
+from AIPscan.models import AIP, FetchJob
 
 SCRIPT_DIR = os.path.dirname(os.path.realpath(__file__))
 FIXTURES_DIR = os.path.join(SCRIPT_DIR, "fixtures")
@@ -176,6 +177,23 @@ def mock_download_mets(
     )
 
 
+def test_delete_fetch_job_task(app_instance, tmpdir, mocker):
+    """Test that fetch job gets deleted by delete fetch job task logic."""
+    storage_service = test_helpers.create_test_storage_service()
+
+    # Create fetch job and confirm expected ID
+    fetch_job1 = test_helpers.create_test_fetch_job(
+        storage_service_id=storage_service.id
+    )
+
+    assert fetch_job1.id == 1
+
+    # Delete fetch job and confirm it no longer exists
+    delete_fetch_job(fetch_job1.id)
+
+    assert FetchJob.query.filter_by(id=fetch_job1.id).first() is None
+
+
 @pytest.mark.parametrize(
     "response, raises_task_error",
     [

AIPscan/Aggregator/views.py

@@ -232,13 +232,10 @@ def new_fetch_job(id):
 
 @aggregator.route("/delete_fetch_job/<id>", methods=["GET"])
 def delete_fetch_job(id):
+    tasks.delete_fetch_job.delay(id)
     fetch_job = FetchJob.query.get(id)
     storage_service = StorageService.query.get(fetch_job.storage_service_id)
-    if os.path.exists(fetch_job.download_directory):
-        shutil.rmtree(fetch_job.download_directory)
-    db.session.delete(fetch_job)
-    db.session.commit()
-    flash("Fetch job {} is deleted".format(fetch_job.download_start))
+    flash("Fetch job {} is being deleted".format(fetch_job.download_start))
     return redirect(url_for("aggregator.storage_service", id=storage_service.id))