(index_info, changes, doc_transform=None, error_collector=None)
return payload
Builds bulk payload json to be called via Elasticsearch Bulk API
Builds bulk payload json to be called via Elasticsearch Bulk API
Builds bulk payload json to be called via Elasticsearch Bulk API
def build_bulk_payload(index_info, changes, doc_transform=None, error_collector=None):
    """
    Builds bulk payload json to be called via Elasticsearch Bulk API
    """
    from corehq.apps.change_feed.document_types import get_doc_meta_object_from_document

    doc_transform = doc_transform or (lambda x: x)
    payload = []

    def _is_deleted(change):
        doc = change.get_document()
        if doc and doc.get('doc_type'):
            return get_doc_meta_object_from_document(doc).is_deletion
        elif change.deleted:
            return bool(change.id)

    for change in changes:
        action = {
            "_index": index_info.alias,
            "_id": change.id
        }
        if settings.ELASTICSEARCH_MAJOR_VERSION == 2:
            action.update({"_type": index_info.type})

        if _is_deleted(change):
            action.update({"_op_type": "delete"})
        elif not change.deleted:
            try:
                doc = change.get_document()
                doc = doc_transform(doc)
                action.update({
                    "_op_type": "index",
                    "_id": doc['_id'],
                    "_source": doc
                })
            except Exception as e:
                if not error_collector:
                    raise
                error_collector.add_error(ChangeError(change, e))
                continue
        payload.append(action)

    return payload
def build_bulk_payload ( index_info , changes , doc_transform = None , error_collector = None ) : from corehq . apps . change_feed . document_types import get_doc_meta_object_from_document doc_transform = doc_transform or ( lambda x : x ) payload = [ ] def _is_deleted ( change ) : doc = change . get_document ( ) if doc and doc . get ( 'doc_type' ) : return get_doc_meta_object_from_document ( doc ) . is_deletion elif change . deleted : return bool ( change . id ) for change in changes : action = { "_index" : index_info . alias , "_id" : change . id } if settings . ELASTICSEARCH_MAJOR_VERSION == 2 : action . update ( { "_type" : index_info . type } ) if _is_deleted ( change ) : action . update ( { "_op_type" : "delete" } ) elif not change . deleted : try : doc = change . get_document ( ) doc = doc_transform ( doc ) action . update ( { "_op_type" : "index" , "_id" : doc [ '_id' ] , "_source" : doc } ) except Exception as e : if not error_collector : raise error_collector . add_error ( ChangeError ( change , e ) ) continue payload . append ( action ) return payload