Merge pull request #515 from DDMAL/celery-debug

Celery debug

requirements.txt

@@ -1,5 +1,5 @@
 backports.ssl-match-hostname==3.4.0.2
-celery==4.4.2
+celery==3.1.25
 certifi==14.05.14
 django-cors-headers==2.4.0
 django-decorator-include==1.3

rodan/celery.py

@@ -8,6 +8,11 @@
 os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'rodan.settings')
 django.setup()
 
+app = Celery('rodan')
+app.config_from_object('django.conf:settings')
+app.autodiscover_tasks(lambda: settings.INSTALLED_APPS)
+
+
 from rodan.jobs.core import (  # noqa
     create_resource,
     create_workflowrun,
@@ -19,9 +24,6 @@
 )
 from rodan.jobs.master_task import master_task  # noqa
 
-app = Celery('rodan')
-app.config_from_object('django.conf:settings')
-app.autodiscover_tasks(lambda: settings.INSTALLED_APPS)
 
 # Core Rodan Tasks
 app.tasks.register(create_resource())

rodan/jobs/core.py

@@ -37,6 +37,7 @@
 from rodan.jobs.base import TemporaryDirectory
 from rodan.jobs.diva_generate_json import GenerateJson
 from rodan.jobs.resource_identification import fileparse
+# from rodan.celery import app
 
 
 class create_resource(Task):
@@ -551,22 +552,33 @@ def convert_string_to_model_dict(dict_):
             """
             Passing messages to celery is faster when using JSON, the problem is you are
             limited by types. Pickle objects are insecure and slower, but the fact that we
-            are searching for the resource again would also slow down things.
+            are searching for the resource again would slow down things.
 
             [TODO] Refactor so we don't need to query the database so often, and look
             into the possibility of using YAML (still faster than pickle but has more
-            options than JSON.)
+            options than JSON. It might be possible to pass a model in yaml and that
+            would make be great.)
             """
             temp_dict = {}
             for item in dict_.items():
                 if isinstance(item[1], list):
-                    temp_dict[InputPort.objects.filter(uuid=item[0])] = [
-                        Resource.objects.filter(uuid=x) for x in item[1]
-                    ]
+                    try:
+                        temp_dict[InputPort.objects.get(uuid=item[0])] = [
+                            Resource.objects.get(uuid=x) for x in item[1]
+                        ]
+                    except Resource.DoesNotExist:
+                        temp_dict[InputPort.objects.get(uuid=item[0])] = [
+                            ResourceList.objects.get(uuid=x) for x in item[1]
+                        ]
                 elif isinstance(item[1], str):
-                    temp_dict[
-                        InputPort.objects.filter(uuid=item[0])[0]
-                    ] = Resource.objects.filter(uuid=item[1][0])
+                    try:
+                        temp_dict[InputPort.objects.get(uuid=item[0])] = Resource.objects.get(
+                            uuid=item[1]
+                        )
+                    except Resource.DoesNotExist:
+                        temp_dict[InputPort.objects.get(uuid=item[0])] = ResourceList.objects.get(
+                            uuid=item[1]
+                        )
                 else:
                     raise Exception(
                         "Unusual input to convert_string_to_model_dict: {}".format(dict_)
@@ -954,3 +966,9 @@ def inner_redo(rj):
 def send_email(subject, body, to):
     email = EmailMessage(subject, body, settings.EMAIL_HOST_USER, to)
     email.send()
+
+
+# app.tasks.register(create_resource())
+# app.tasks.register(package_results())
+# app.tasks.register(expire_package)
+# app.tasks.register(create_workflowrun())

rodan/jobs/load.py

@@ -429,7 +429,7 @@ def set_version(module):
 
 if job_list:  # there are database jobs that are not registered. Should delete them.
     # To keep docker images small, only the main celery queue NEEDS all jobs.
-    if os.environ["CELERY_JOB_QUEUE"] != "celery":
+    if os.environ["CELERY_JOB_QUEUE"] != "celery" and os.environ["CELERY_JOB_QUEUE"] != "None":
         pass
     elif not UPDATE_JOBS:
         raise ImproperlyConfigured(