(
        pillow_id='case-pillow', ucr_division=None,
        include_ucrs=None, exclude_ucrs=None,
        num_processes=1, process_num=0, ucr_configs=None, skip_ucr=False,
        processor_chunk_size=DEFAULT_PROCESSOR_CHUNK_SIZE, topics=None,
        dedicated_migration_process=False, **kwargs)
return ConstructedPillow(
        name=pillow_id,
        change_feed=change_feed,
        checkpoint=checkpoint,
        change_processed_event_handler=event_handler,
        processor=processors,
        processor_chunk_size=processor_chunk_size,
        process_num=process_num,
        is_dedicated_migration_process=dedicated_migration_process and run_migrations
    )
Return a pillow that processes cases. The processors include, UCR and elastic processors

    Processors:
      - :py:class:`corehq.apps.userreports.pillow.ConfigurableReportPillowProcessor` (disabled when skip_ucr=True)
      - :py:class:`pillowtop.processors.elastic.BulkElasticProcessor`
      - :py:func:`corehq.pillows.case_search.get_case_search_processor`
      - :py:class:`corehq.messaging.pillow.CaseMessagingSyncProcessor`
Return a pillow that processes cases. The processors include, UCR and elastic processors
Return a pillow that processes cases . The processors include UCR and elastic processors
def get_case_pillow(
        pillow_id='case-pillow', ucr_division=None,
        include_ucrs=None, exclude_ucrs=None,
        num_processes=1, process_num=0, ucr_configs=None, skip_ucr=False,
        processor_chunk_size=DEFAULT_PROCESSOR_CHUNK_SIZE, topics=None,
        dedicated_migration_process=False, **kwargs):
    """Return a pillow that processes cases. The processors include, UCR and elastic processors

    Processors:
      - :py:class:`corehq.apps.userreports.pillow.ConfigurableReportPillowProcessor` (disabled when skip_ucr=True)
      - :py:class:`pillowtop.processors.elastic.BulkElasticProcessor`
      - :py:func:`corehq.pillows.case_search.get_case_search_processor`
      - :py:class:`corehq.messaging.pillow.CaseMessagingSyncProcessor`
    """
    if topics:
        assert set(topics).issubset(CASE_TOPICS), "This is a pillow to process cases only"
    topics = topics or CASE_TOPICS
    change_feed = KafkaChangeFeed(
        topics, client_id=pillow_id, num_processes=num_processes, process_num=process_num,
        dedicated_migration_process=dedicated_migration_process
    )
    run_migrations = (process_num == 0)  # only first process runs migrations
    ucr_processor = get_ucr_processor(
        data_source_providers=[
            DynamicDataSourceProvider('CommCareCase'),
            StaticDataSourceProvider('CommCareCase')
        ],
        ucr_division=ucr_division,
        include_ucrs=include_ucrs,
        exclude_ucrs=exclude_ucrs,
        run_migrations=run_migrations,
        ucr_configs=ucr_configs
    )
    ucr_dr_processor = get_data_registry_ucr_processor(
        run_migrations=run_migrations,
        ucr_configs=ucr_configs
    )
    case_to_es_processor = BulkElasticProcessor(
        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
    )
    case_search_processor = get_case_search_processor()

    checkpoint_id = "{}-{}-{}-{}".format(
        pillow_id, CASE_INDEX_INFO.index, case_search_processor.index_info.index, 'messaging-sync')
    checkpoint = KafkaPillowCheckpoint(checkpoint_id, topics)
    event_handler = KafkaCheckpointEventHandler(
        checkpoint=checkpoint, checkpoint_frequency=1000, change_feed=change_feed,
        checkpoint_callback=ucr_processor
    )
    processors = [case_to_es_processor, CaseMessagingSyncProcessor()]
    if settings.RUN_CASE_SEARCH_PILLOW:
        processors.append(case_search_processor)
    if not settings.ENTERPRISE_MODE:
        processors.append(get_case_to_report_es_processor())
    if not skip_ucr:
        # this option is useful in tests to avoid extra UCR setup where unneccessary
        processors = [ucr_processor, ucr_dr_processor] + processors
    return ConstructedPillow(
        name=pillow_id,
        change_feed=change_feed,
        checkpoint=checkpoint,
        change_processed_event_handler=event_handler,
        processor=processors,
        processor_chunk_size=processor_chunk_size,
        process_num=process_num,
        is_dedicated_migration_process=dedicated_migration_process and run_migrations
    )
def get_case_pillow ( pillow_id = 'case-pillow' , ucr_division = None , include_ucrs = None , exclude_ucrs = None , num_processes = 1 , process_num = 0 , ucr_configs = None , skip_ucr = False , processor_chunk_size = DEFAULT_PROCESSOR_CHUNK_SIZE , topics = None , dedicated_migration_process = False , * * kwargs ) : if topics : assert set ( topics ) . issubset ( CASE_TOPICS ) , "This is a pillow to process cases only" topics = topics or CASE_TOPICS change_feed = KafkaChangeFeed ( topics , client_id = pillow_id , num_processes = num_processes , process_num = process_num , dedicated_migration_process = dedicated_migration_process ) run_migrations = ( process_num == 0 ) # only first process runs migrations ucr_processor = get_ucr_processor ( data_source_providers = [ DynamicDataSourceProvider ( 'CommCareCase' ) , StaticDataSourceProvider ( 'CommCareCase' ) ] , ucr_division = ucr_division , include_ucrs = include_ucrs , exclude_ucrs = exclude_ucrs , run_migrations = run_migrations , ucr_configs = ucr_configs ) ucr_dr_processor = get_data_registry_ucr_processor ( run_migrations = run_migrations , ucr_configs = ucr_configs ) case_to_es_processor = BulkElasticProcessor ( 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 ) case_search_processor = get_case_search_processor ( ) checkpoint_id = "{}-{}-{}-{}" . format ( pillow_id , CASE_INDEX_INFO . index , case_search_processor . index_info . index , 'messaging-sync' ) checkpoint = KafkaPillowCheckpoint ( checkpoint_id , topics ) event_handler = KafkaCheckpointEventHandler ( checkpoint = checkpoint , checkpoint_frequency = 1000 , change_feed = change_feed , checkpoint_callback = ucr_processor ) processors = [ case_to_es_processor , CaseMessagingSyncProcessor ( ) ] if settings . RUN_CASE_SEARCH_PILLOW : processors . append ( case_search_processor ) if not settings . ENTERPRISE_MODE : processors . append ( get_case_to_report_es_processor ( ) ) if not skip_ucr : # this option is useful in tests to avoid extra UCR setup where unneccessary processors = [ ucr_processor , ucr_dr_processor ] + processors return ConstructedPillow ( name = pillow_id , change_feed = change_feed , checkpoint = checkpoint , change_processed_event_handler = event_handler , processor = processors , processor_chunk_size = processor_chunk_size , process_num = process_num , is_dedicated_migration_process = dedicated_migration_process and run_migrations )