tasks.py

# -*- coding: UTF-8 -*-

from celery import shared_task
from django.conf import settings
from django.core import serializers
from trans2es.type_info import get_type_info_map
from rpc.all import get_rpc_remote_invoker
from libs.es import ESPerform

@shared_task
def write_to_es(es_type, pk_list, configuration, use_batch_query_set=False):
    pk_list = list(frozenset(pk_list))
    type_info_map = get_type_info_map()
    type_info = type_info_map[es_type]
    type_info.insert_table_by_pk_list(
        sub_index_name=es_type,
        pk_list=pk_list,
        use_batch_query_set=use_batch_query_set,
        es=ESPerform.get_cli()
    )