Merge pull request #742 from Bidaya0/expirement/add-celery-handler

Expirement/add celery handler

dongtai_common/utils/es.py

@@ -0,0 +1,26 @@
+from celery import shared_task
+from django.apps import apps
+from django.db import transaction
+from django_elasticsearch_dsl.registries import registry
+from django_elasticsearch_dsl.signals import RealTimeSignalProcessor
+import logging
+
+
+logger = logging.getLogger('dongtai-core')
+
+@shared_task
+def handle_save(pk, app_label, model_name):
+    logger.info(f'handle_save to es: {model_name} pk: {pk}')
+    sender = apps.get_model(app_label, model_name)
+    instance = sender.objects.get(pk=pk)
+    registry.update(instance)
+    registry.update_related(instance)
+
+
+class DTCelerySignalProcessor(RealTimeSignalProcessor):
+
+    def handle_save(self, sender, instance, **kwargs):
+        app_label = instance._meta.app_label
+        model_name = instance._meta.model_name
+        transaction.on_commit(
+            lambda: handle_save.delay(instance.pk, app_label, model_name))

dongtai_conf/settings.py

@@ -609,6 +609,8 @@ def safe_execute(default, exception, function, *args):
     ASSET_INDEX = config.get('elastic_search', 'asset_index')
     ELASTICSEARCH_DSL_PARALLEL = True
     ELASTICSEARCH_DSL_AUTO_REFRESH = False
+    ELASTICSEARCH_DSL_AUTO_REFRESH = False
+    ELASTICSEARCH_DSL_SIGNAL_PROCESSOR = 'dongtai_common.utils.es.DTCelerySignalProcessor'
     from elasticsearch import logger as es_logger
     import elasticsearch
     es_logger.setLevel(elasticsearch.logging.INFO)