Celery: manage cached_label update with tasks

2 files changed, 72 insertions, 9 deletions

diff --git a/ishtar_common/models_imports.py b/ishtar_common/models_imports.py
index 904f3f013..20c3bf9c4 100644
--- a/ishtar_common/models_imports.py
+++ b/ishtar_common/models_imports.py
@@ -47,11 +47,7 @@ from ishtar_common.data_importer import Importer, ImportFormater, \
     IntegerFormater, FloatFormater, UnicodeFormater, DateFormater, \
     TypeFormater, YearFormater, StrToBoolean, FileFormater, InseeFormater, \
     ImporterError
-
-if settings.USE_BACKGROUND_TASK:
-    from background_task import background
-else:
-    background = lambda x: x
+from ishtar_common.utils import task
 
 logger = logging.getLogger(__name__)
 
@@ -859,7 +855,7 @@ delayed_check = None
 
 
 if settings.USE_BACKGROUND_TASK:
-    @background(schedule=1)
+    @task()
     def delayed_import(import_pk, session_key):
         try:
             imp = Import.objects.get(pk=import_pk)
@@ -867,7 +863,7 @@ if settings.USE_BACKGROUND_TASK:
             pass
         imp.importation(session_key=session_key)
 
-    @background(schedule=1)
+    @task()
     def delayed_check(import_pk, session_key):
         try:
             imp = Import.objects.get(pk=import_pk)
diff --git a/ishtar_common/utils.py b/ishtar_common/utils.py
index d0bfee02a..55301af14 100644
--- a/ishtar_common/utils.py
+++ b/ishtar_common/utils.py
@@ -62,10 +62,22 @@ def dict_to_tuple(dct):
 
 
 def debug_line_no():
-    cf = currentframe()
     return currentframe().f_back.f_lineno
 
 
+def fake_task(*args):
+    def fake(func):
+        return func
+    return fake
+
+
+if settings.USE_BACKGROUND_TASK:
+    from celery import shared_task
+    task = shared_task
+else:
+    task = fake_task
+
+
 class BColors:
     """
     Bash colors. Don't forget to finish your colored string with ENDC.
@@ -312,6 +324,28 @@ def force_cached_label_changed(sender, **kwargs):
     cached_label_changed(sender, **kwargs)
 
 
+def serialize_args_for_tasks(sender, **kwargs):
+    if 'signal' in kwargs:
+        kwargs.pop('signal')
+    if 'instance' in kwargs:
+        kwargs['instance'] = kwargs["instance"].pk
+    sender = (sender._meta.app_label, sender._meta.object_name)
+    return sender, kwargs
+
+
+def deserialize_args_for_tasks(sender, **kwargs):
+    if "instance" not in kwargs:
+        return sender, None
+    if not isinstance(sender, (tuple, list)):  # not task
+        return sender, kwargs["instance"]
+    sender = apps.get_model(*sender)
+    try:
+        instance = sender.objects.get(pk=kwargs['instance'])
+    except sender.DoesNotExist:
+        return sender, None  # object is not here anymore
+    return sender, instance
+
+
 def cached_label_changed(sender, **kwargs):
     if not kwargs.get('instance'):
         return
@@ -324,9 +358,25 @@ def cached_label_changed(sender, **kwargs):
     if not force_update and hasattr(instance, '_cached_label_checked') \
             and instance._cached_label_checked:
         return
+    if not settings.USE_BACKGROUND_TASK:
+        return _cached_label_changed(sender, **kwargs)
+
+    sender, kwargs = serialize_args_for_tasks(sender, **kwargs)
+    return _cached_label_changed.delay(sender, **kwargs)
+
+
+@task()
+def _cached_label_changed(sender, **kwargs):
+    sender, instance = deserialize_args_for_tasks(sender, **kwargs)
+    if not instance:
+        return
+
+    force_update = kwargs.get('force_update', False)
+
     if hasattr(instance, "refresh_cache"):
         instance.refresh_cache()
 
+    # print(sender, "cached_label_changed")
     instance._cached_label_checked = True
     cached_labels = ['cached_label']
     if hasattr(sender, 'CACHED_LABELS'):
@@ -505,19 +555,36 @@ def post_save_geo(sender, **kwargs):
     if not kwargs.get('instance') or getattr(kwargs['instance'],
                                              '_post_saved_geo', False):
         return
+
+    if not settings.USE_BACKGROUND_TASK:
+        return _post_save_geo(sender, **kwargs)
+    sender, kwargs = serialize_args_for_tasks(sender, **kwargs)
+    return _post_save_geo.delay(sender, **kwargs)
+
+
+@task()
+def _post_save_geo(sender, **kwargs):
+    """
+    Convert raw x, y, z point to real geo field
+    """
     from ishtar_common.models import get_current_profile  # not clean but utils
     # is loaded before models
 
     profile = get_current_profile()
     if not profile.mapping:
         return
-    instance = kwargs.get('instance')
+    sender, instance = deserialize_args_for_tasks(sender, **kwargs)
+    if not instance:
+        return
+
     kls_name = instance.__class__.__name__
 
     if not profile.locate_warehouses and ("Container" in kls_name
                                           or "Warehouse" in kls_name):
         return
 
+    # print(sender, "post_save_geo")
+
     current_source = "default"
     if hasattr(instance.__class__, "_meta"):
         current_source = str(instance.__class__._meta.verbose_name)