(pillow_id='CaseToElasticsearchPillow', num_processes=1,
                                     process_num=0, **kwargs)
return ConstructedPillow(
        name=pillow_id,
        checkpoint=checkpoint,
        change_feed=kafka_change_feed,
        processor=case_processor,
        change_processed_event_handler=KafkaCheckpointEventHandler(
            checkpoint=checkpoint, checkpoint_frequency=100, change_feed=kafka_change_feed
        ),
    )
Return a pillow that processes cases to Elasticsearch.

    Processors:
      - :py:class:`pillowtop.processors.elastic.ElasticProcessor`
Return a pillow that processes cases to Elasticsearch.
Return a pillow that processes cases to Elasticsearch .
def get_case_to_elasticsearch_pillow(pillow_id='CaseToElasticsearchPillow', num_processes=1,
                                     process_num=0, **kwargs):
    """Return a pillow that processes cases to Elasticsearch.

    Processors:
      - :py:class:`pillowtop.processors.elastic.ElasticProcessor`
    """
    # todo; To remove after full rollout of https://github.com/dimagi/commcare-hq/pull/21329/
    assert pillow_id == 'CaseToElasticsearchPillow', 'Pillow ID is not allowed to change'
    checkpoint = get_checkpoint_for_elasticsearch_pillow(pillow_id, CASE_INDEX_INFO, CASE_TOPICS)
    case_processor = ElasticProcessor(
        elasticsearch=get_es_new(),
        index_info=CASE_INDEX_INFO,
        doc_prep_fn=transform_case_for_elasticsearch,
        change_filter_fn=is_couch_change_for_sql_domain
    )
    kafka_change_feed = KafkaChangeFeed(
        topics=CASE_TOPICS, client_id='cases-to-es', num_processes=num_processes, process_num=process_num
    )
    return ConstructedPillow(
        name=pillow_id,
        checkpoint=checkpoint,
        change_feed=kafka_change_feed,
        processor=case_processor,
        change_processed_event_handler=KafkaCheckpointEventHandler(
            checkpoint=checkpoint, checkpoint_frequency=100, change_feed=kafka_change_feed
        ),
    )
def get_case_to_elasticsearch_pillow ( pillow_id = 'CaseToElasticsearchPillow' , num_processes = 1 , process_num = 0 , * * kwargs ) : # todo; To remove after full rollout of https://github.com/dimagi/commcare-hq/pull/21329/ assert pillow_id == 'CaseToElasticsearchPillow' , 'Pillow ID is not allowed to change' checkpoint = get_checkpoint_for_elasticsearch_pillow ( pillow_id , CASE_INDEX_INFO , CASE_TOPICS ) case_processor = ElasticProcessor ( elasticsearch = get_es_new ( ) , index_info = CASE_INDEX_INFO , doc_prep_fn = transform_case_for_elasticsearch , change_filter_fn = is_couch_change_for_sql_domain ) kafka_change_feed = KafkaChangeFeed ( topics = CASE_TOPICS , client_id = 'cases-to-es' , num_processes = num_processes , process_num = process_num ) return ConstructedPillow ( name = pillow_id , checkpoint = checkpoint , change_feed = kafka_change_feed , processor = case_processor , change_processed_event_handler = KafkaCheckpointEventHandler ( checkpoint = checkpoint , checkpoint_frequency = 100 , change_feed = kafka_change_feed ) , )