---
# System prepended metadata

title: FLUSSO DEL NEW INDEX NAME

---

# FLUSSO DEL NEW INDEX NAME

il metodo **esclient.build_insert_bulk()** viene usato nel metodo **thing_index.insert_bulk_reindex** alla **riga 336**.

    def insert_bulk_reindex(self, new_things_index_name, things_document_list):  
	    body = self.es_client.build_insert_bulk({new_things_index_name: things_document_list})  
	    results = self.es_client.bulk(body)  
	    return results
nel quale gli viene passato il nuovo index.
Esso viene usato in **reindex/lambdafunction.py nel metodo fill_new_indexes()** alla riga **434.**

    def fill_new_indexes(new_things_index_name, document_list):  
    errors = []  
    if len(document_list) == 0:  
        return errors  
    results = things_index.insert_bulk_reindex(new_things_index_name, document_list)  
    for item in results['items']:  
        if item['index']['status'] != 201:  
            if item['index']['status'] != 200:  
                errors.append(  
                    'ES error in index {}: {}'.format(item["index"]["_index"], item["index"]['error']['reason']))  
            # else:
            #errors.append('Warn: Insert operation updated an object that was already present!')
            return errors
              
il **metodo** suddetto viene poi utilizzato nel metodo **reindex** della stessa lamda alla riga **181**

    def reindex(event, context):  
        if 'uuid_source_mapping' not in event and \  
                'skip_adjust_alias' not in event and \  
                'skip_stop_lambda_event_sources' not in event:  
            # stop the parsing of new events  
      lambda_sources = {  
                SQS_STREAM_ARN: DD_PARSER_LAMBDA_NAME,  
      KINESIS_STREAM_ARN: ES_PARSER_LAMBDA_NAME  
            }  
            uuid_source_mapping = stop_lambda_event_sources(lambda_sources)  
        else:  
            uuid_source_mapping = event.get('uuid_source_mapping')  
      
        if 'new_things_index_name' not in event:  
            new_things_index_name = things_index.create()  
        else:  
            new_things_index_name = event['new_things_index_name']  
      
        if 'project_ids' in event:  
            project_ids = event['project_ids']  
            logger.debug(f"Projects to scan retrieved from the event: {project_ids}")  
        else:  
            project_ids = projects_details_table.get_project_ids()  
            logger.debug(f"Projects to scan retrieved from DynamoDB: {project_ids}")  
            if DROP_LAMBDA_LOGS:  
                lambdas_index.drop()  
      
        n_projects_to_query = len(project_ids)  
        project_ids_to_pass = deepcopy(project_ids)  
        errors = []  
      
        things_next_token: Optional[str] = None  
      gateways_next_token: Optional[str] = None  
      
      # check whether there's a project to complete  
      if 'things_next_token' in event:  
            things_next_token = event['things_next_token']  
            del event['things_next_token']  
        if 'gateways_next_token' in event:  
            gateways_next_token = event['gateways_next_token']  
            del event['gateways_next_token']  
      
        is_first_cycle: bool = things_next_token is None and gateways_next_token is None  
      
      # main loop for the list of projects  
      for project_id in project_ids:  
      
            # main loop for current project  
      while is_first_cycle or things_next_token is not None or gateways_next_token is not None:  
      
                try:  
                    # invokes the same lambda asynchronously with remaining project ids to prevent timeout  
      t = context.get_remaining_time_in_millis()  
      
                    if t < PROCESS_TIMEOUT:  # 30s is the worst case scenario for a call to IoT Core  
     # if no progress have been made from the last call, something is wrong  if n_projects_to_query == len(  
                                project_ids_to_pass) and things_next_token is None and gateways_next_token is None:  
                            logger.error(  
                                "Error during reindex, no project has been elaborated before a new asynchronous call")  
                            return ["Stall Error"]  
      
                        event['project_ids'] = project_ids_to_pass  
                        if things_next_token is not None:  
                            event['things_next_token'] = things_next_token  
                        if gateways_next_token is not None:  
                            event['gateways_next_token'] = gateways_next_token  
      
                        event['uuid_source_mapping'] = uuid_source_mapping  
                        event['new_things_index_name'] = new_things_index_name  
      
                        invoke_async_lambda_response = lambda_client.invoke(  
                            FunctionName=context.function_name,  
      InvocationType='Event',  
      Payload=json.dumps(event)  
                        )  
                        logger.debug(  
                            f"Invoked requestId function"  
     f"{invoke_async_lambda_response['ResponseMetadata']['RequestId']} with payload: {event}"  
      )  
      
                        return errors  
      
                    project_info = projects_details_table.get_project_details_obj(project_id)  
                    things_list = []  
      
                    logger.debug(f"Scanning project: {project_info.to_json()}")  
      
                    if is_first_cycle or gateways_next_token is not None:  
                        devices = get_devices_from_project(project_info, 'gateway', next_token=gateways_next_token)  
                        things_list.extend(devices['list'])  
                        gateways_next_token = devices['next_token']  
      
                    if is_first_cycle or things_next_token is not None:  
                        devices = get_devices_from_project(project_info, 'thing', next_token=things_next_token)  
                        things_list.extend(devices['list'])  
                        things_next_token = devices['next_token']  
      
                    is_first_cycle = False  
      
      errors.extend(fill_new_indexes(new_things_index_name, things_list))  
      
                except Exception as e:  
                    logger.error("Error for project %s: %s", project_id, e)  
                    errors.extend(["Error for project {}: {}".format(project_id, str(e))])  
                    # skip project  
      break  
      
     if 'skip_align_lambda_invoke' not in event:  
                data_to_insert = {  
                    'update_mode': ["status", "certificate"],  
      'things_list': {  
                        project_id: ["*"]  
                    }  
                }  
                try:  
                    lambda_client.invoke(  
                        FunctionName=os.environ['ALIGN_LAMBDA_NAME'],  
      InvocationType='Event',  
      Payload=json.dumps(data_to_insert)  
                    )  
                    logger.debug(f"Launched things alignment task for {project_id}")  
                except Exception as e:  
                    logger.exception("Error invoking lambda align thing for for project %s")  
                    # proceed with the next project  
      continue  
      
      project_ids_to_pass.remove(project_id)  
            things_next_token = None  
      gateways_next_token = None  
      is_first_cycle = True  
      
      # all projects have been processes  
      if 'skip_adjust_alias' not in event:  
            adjust_alias()  
        if uuid_source_mapping:  
            restore_lambda_event_sources(uuid_source_mapping)  
      
        return errors
**new_things_index_name** viene usato alle righe **94,95, 97 e 150**

il mio obiettivo è usare **new_things_index_name** come altro valore da passare in pancia al metodo custom da lanciare quando la reindex è attiva e abbiamo bisogno di aggiornare cancellare e aggiungere i record che ci interessano durante la reindex.