[Fixes #4813] Removing a layer is broken - Celery error (#4817)

geonode/documents/tasks.py

@@ -22,7 +22,7 @@
 from os import access, R_OK
 from os.path import isfile
 
-from celery.app import shared_task
+from geonode.celery_app import app
 from celery.utils.log import get_task_logger
 
 from geonode.documents.models import Document
@@ -34,7 +34,7 @@
 logger = get_task_logger(__name__)
 
 
-@shared_task(bind=True, queue='update')
+@app.task(bind=True, queue='update')
 def create_document_thumbnail(self, object_id):
     """
     Create thumbnail for a document.
@@ -87,13 +87,13 @@ def create_document_thumbnail(self, object_id):
     logger.debug("Thumbnail for document #{} created.".format(object_id))
 
 
-@shared_task(bind=True, queue='cleanup')
+@app.task(bind=True, queue='cleanup')
 def delete_orphaned_document_files(self):
     from geonode.documents.utils import delete_orphaned_document_files
     delete_orphaned_document_files()
 
 
-@shared_task(bind=True, queue='cleanup')
+@app.task(bind=True, queue='cleanup')
 def delete_orphaned_thumbnails(self):
     from geonode.base.utils import delete_orphaned_thumbs
     delete_orphaned_thumbs()

geonode/geoserver/tasks.py

@@ -18,15 +18,15 @@
 #
 #########################################################################
 
-from celery.app import shared_task
+from geonode.celery_app import app
 from celery.utils.log import get_task_logger
 
 from .helpers import gs_slurp
 
 logger = get_task_logger(__name__)
 
 
-@shared_task(bind=True, queue='update')
+@app.task(bind=True, queue='update')
 def geoserver_update_layers(self, *args, **kwargs):
     """
     Runs update layers.

geonode/layers/tasks.py

@@ -20,24 +20,24 @@
 
 """celery tasks for geonode.layers."""
 
-from celery import shared_task
+from geonode.celery_app import app
 from celery.utils.log import get_task_logger
 
 from geonode.layers.models import Layer
 
 logger = get_task_logger(__name__)
 
 
-@shared_task(bind=True,
-             name='geonode.layers.tasks.update.delete_layer',
-             queue='cleanup',
-             autoretry_for=(Layer.DoesNotExist, ),
-             retry_kwargs={'max_retries': 5, 'countdown': 5})
+@app.task(bind=True, queue='cleanup', expires=300)
 def delete_layer(self, layer_id):
     """
     Deletes a layer.
     """
-    layer = Layer.objects.get(id=layer_id)
+
+    try:
+        layer = Layer.objects.get(id=layer_id)
+    except Layer.DoesNotExist:
+        return
+
     logger.info('Deleting Layer {0}'.format(layer))
     layer.delete()
-    return True

geonode/maps/tasks.py

@@ -20,20 +20,19 @@
 
 """celery tasks for geonode.maps."""
 
-from celery.app import shared_task
+from geonode.celery_app import app
 from celery.utils.log import get_task_logger
 
 from geonode.maps.models import Map
 
 logger = get_task_logger(__name__)
 
 
-@shared_task(bind=True, queue='cleanup', expires=300)
+@app.task(bind=True, queue='cleanup', expires=300)
 def delete_map(self, object_id):
     """
     Deletes a map and the associated map layers.
     """
-
     try:
         map_obj = Map.objects.get(id=object_id)
     except Map.DoesNotExist:

geonode/qgis_server/tasks/update.py

@@ -23,7 +23,7 @@
 import socket
 
 import requests
-from celery import shared_task
+from geonode.celery_app import app
 from requests.exceptions import HTTPError
 
 
@@ -39,7 +39,7 @@
 logger = logging.getLogger(__name__)
 
 
-@shared_task(
+@app.task(
     name='geonode.qgis_server.tasks.update.create_qgis_server_thumbnail',
     queue='update',
     autoretry_for=(QGISServerLayer.DoesNotExist, ),
@@ -109,7 +109,7 @@ def create_qgis_server_thumbnail(instance, overwrite=False, bbox=None):
         return False
 
 
-@shared_task(
+@app.task(
     name='geonode.qgis_server.tasks.update.cache_request',
     queue='update')
 @on_ogc_backend(qgis_server.BACKEND_PACKAGE)

geonode/services/tasks.py

@@ -23,22 +23,22 @@
 
 import logging
 
-from celery import shared_task
 from django.db import transaction
 
 from . import enumerations
 from . import models
 from .serviceprocessors import get_service_handler
 
+from geonode.celery_app import app
 from geonode.layers.models import Layer
 from geonode.catalogue.models import catalogue_post_save
 
 logger = logging.getLogger(__name__)
 
 
-@shared_task(bind=True,
-             name='geonode.services.tasks.update.harvest_resource',
-             queue='update',)
+@app.task(bind=True,
+          name='geonode.services.tasks.update.harvest_resource',
+          queue='update',)
 def harvest_resource(self, harvest_job_id):
     harvest_job = models.HarvestJob.objects.get(pk=harvest_job_id)
     harvest_job.update_status(

geonode/settings.py

@@ -1565,10 +1565,14 @@ def get_geonode_catalogue_service():
 if ASYNC_SIGNALS:
     _BROKER_URL = os.environ.get('BROKER_URL', RABBITMQ_SIGNALS_BROKER_URL)
     # _BROKER_URL =  = os.environ.get('BROKER_URL', REDIS_SIGNALS_BROKER_URL)
-
     CELERY_RESULT_BACKEND = _BROKER_URL
 else:
     _BROKER_URL = LOCAL_SIGNALS_BROKER_URL
+    CELERY_RESULT_BACKEND_PATH = os.getenv(
+        'CELERY_RESULT_BACKEND_PATH', os.path.join(PROJECT_ROOT, 'results'))
+    if not os.path.exists(CELERY_RESULT_BACKEND_PATH):
+        os.makedirs(CELERY_RESULT_BACKEND_PATH)
+    CELERY_RESULT_BACKEND = 'file:///%s' % CELERY_RESULT_BACKEND_PATH
 
 # Note:BROKER_URL is deprecated in favour of CELERY_BROKER_URL
 CELERY_BROKER_URL = _BROKER_URL
@@ -1580,6 +1584,7 @@ def get_geonode_catalogue_service():
 
 # Set this to False in order to run async
 CELERY_TASK_ALWAYS_EAGER = False if ASYNC_SIGNALS else True
+CELERY_TASK_EAGER_PROPAGATES = False if ASYNC_SIGNALS else True
 CELERY_TASK_IGNORE_RESULT = True
 
 # I use these to debug kombu crashes; we get a more informative message.

geonode/tasks/tasks.py

@@ -18,17 +18,17 @@
 #
 #########################################################################
 
-from celery.app import shared_task
+from geonode.celery_app import app
 from celery.utils.log import get_task_logger
 from django.conf import settings
 from django.core.mail import send_mail
 
 logger = get_task_logger(__name__)
 
 
-@shared_task(bind=True,
-             name='geonode.tasks.email.send_mail',
-             queue='email',)
+@app.task(bind=True,
+          name='geonode.tasks.email.send_mail',
+          queue='email',)
 def send_email(self, *args, **kwargs):
     """
     Sends an email using django's send_mail functionality.
@@ -37,9 +37,9 @@ def send_email(self, *args, **kwargs):
     send_mail(*args, **kwargs)
 
 
-@shared_task(bind=True,
-             name='geonode.tasks.notifications.send_queued_notifications',
-             queue='email',)
+@app.task(bind=True,
+          name='geonode.tasks.notifications.send_queued_notifications',
+          queue='email',)
 def send_queued_notifications(self, *args):
     """Sends queued notifications.