parameters: (processor, bulk_objs, bulk_ser_objs,
                         raise_on_exception=True, raise_on_error=True,
                         *args, **kwargs)
docstring: Send a bulk request to xapiand and process the output.
docstring_summary: Send a bulk request to xapiand and process the output.
docstring_tokens: Send a bulk request to xapiand and process the output .
function: def process_bulk_chunk(processor, bulk_objs, bulk_ser_objs,
                         raise_on_exception=True, raise_on_error=True,
                         *args, **kwargs):
      """
      Send a bulk request to xapiand and process the output.
      """
  
      try:
          # send the actual request
          resp = processor(bulk_ser_objs, *args, **kwargs)
      except TransportError as e:
          # default behavior - just propagate exception
          if raise_on_exception:
              raise e
  
          # if we are not propagating, mark all objects in current chunk as failed
          err_message = str(e)
          exc_errors = []
  
          for obj in bulk_objs:
              # collect all the information about failed objects
              info = {
                  '_status': e.status_code,
                  '_message': [err_message],
                  '_exception': e,
              }
              info.update(obj)
              exc_errors.append(info)
  
          # emulate standard behavior for failed objects
          if raise_on_error:
              raise BulkIndexError("%d document(s) failed to index." % len(exc_errors), exc_errors)
          else:
              status = e.status_code
              for err in exc_errors:
                  yield status, err
              return
  
      # if raise on error is set, we need to collect errors per chunk before raising them
      errors = []
  
      # go through request-response pairs and detect failures
      for obj, item in zip(bulk_objs, resp['items']):
          status = item.get('_status', 200)
          # include original document source
          item.update(obj)
          if 200 <= status < 300 or not raise_on_error:
              # if we are not just recording all errors to be able to raise
              # them all at once, yield items individually
              yield status, item
          else:
              errors.append(item)
  
      if errors:
          raise BulkIndexError("%d document(s) failed to index." % len(errors), errors)
function_tokens: def process_bulk_chunk ( processor , bulk_objs , bulk_ser_objs , raise_on_exception = True , raise_on_error = True , * args , * * kwargs ) : try : # send the actual request resp = processor ( bulk_ser_objs , * args , * * kwargs ) except TransportError as e : # default behavior - just propagate exception if raise_on_exception : raise e # if we are not propagating, mark all objects in current chunk as failed err_message = str ( e ) exc_errors = [ ] for obj in bulk_objs : # collect all the information about failed objects info = { '_status' : e . status_code , '_message' : [ err_message ] , '_exception' : e , } info . update ( obj ) exc_errors . append ( info ) # emulate standard behavior for failed objects if raise_on_error : raise BulkIndexError ( "%d document(s) failed to index." % len ( exc_errors ) , exc_errors ) else : status = e . status_code for err in exc_errors : yield status , err return # if raise on error is set, we need to collect errors per chunk before raising them errors = [ ] # go through request-response pairs and detect failures for obj , item in zip ( bulk_objs , resp [ 'items' ] ) : status = item . get ( '_status' , 200 ) # include original document source item . update ( obj ) if 200 <= status < 300 or not raise_on_error : # if we are not just recording all errors to be able to raise # them all at once, yield items individually yield status , item else : errors . append ( item ) if errors : raise BulkIndexError ( "%d document(s) failed to index." % len ( errors ) , errors )