parameters: (processor, objects, serializer,
                     chunk_size=500, max_chunk_bytes=100 * 1024 * 1024,
                     raise_on_error=True, raise_on_exception=True, max_retries=0,
                     initial_backoff=2, max_backoff=600, yield_ok=True, *args, **kwargs)
docstring: Streaming bulk consumes objects from the iterable passed in and yields
      results per object. For non-streaming usecases use
      :func:`~xapiand.helpers.bulk` which is a wrapper around streaming
      bulk that returns summary information about the bulk operation once the
      entire input is consumed and sent.
  
      If you specify ``max_retries`` it will also retry any documents that were
      rejected with a ``429`` status code. To do this it will wait (**by calling
      time.sleep which will block**) for ``initial_backoff`` seconds and then,
      every subsequent rejection for the same chunk, for double the time every
      time up to ``max_backoff`` seconds.
  
      :arg processor: function to process the objects chunks
      :arg objects: iterable containing the objects to be processed
      :arg serializer: serializer instance
      :arg chunk_size: number of docs in one chunk sent to the server (default: 500)
      :arg max_chunk_bytes: the maximum size of the request in bytes (default: 100MB)
      :arg raise_on_error: raise ``BulkIndexError`` containing errors (as `.errors`)
          from the execution of the last chunk when some occur. By default we raise.
      :arg raise_on_exception: if ``False`` then don't propagate exceptions from
          call to ``bulk`` and just report the items that failed as failed.
      :arg max_retries: maximum number of times a document will be retried when
          ``429`` is received, set to 0 (default) for no retries on ``429``
      :arg initial_backoff: number of seconds we should wait before the first
          retry. Any subsequent retries will be powers of ``initial_backoff *
          2**retry_number``
      :arg max_backoff: maximum number of seconds a retry will wait
      :arg yield_ok: if set to False will skip successful documents in the output
docstring_summary: Streaming bulk consumes objects from the iterable passed in and yields
      results per object. For non-streaming usecases use
      :func:`~xapiand.helpers.bulk` which is a wrapper around streaming
      bulk that returns summary information about the bulk operation once the
      entire input is consumed and sent.
docstring_tokens: Streaming bulk consumes objects from the iterable passed in and yields results per object . For non - streaming usecases use : func : ~xapiand . helpers . bulk which is a wrapper around streaming bulk that returns summary information about the bulk operation once the entire input is consumed and sent .
function: def streaming_bulk(processor, objects, serializer,
                     chunk_size=500, max_chunk_bytes=100 * 1024 * 1024,
                     raise_on_error=True, raise_on_exception=True, max_retries=0,
                     initial_backoff=2, max_backoff=600, yield_ok=True, *args, **kwargs):
      """
      Streaming bulk consumes objects from the iterable passed in and yields
      results per object. For non-streaming usecases use
      :func:`~xapiand.helpers.bulk` which is a wrapper around streaming
      bulk that returns summary information about the bulk operation once the
      entire input is consumed and sent.
  
      If you specify ``max_retries`` it will also retry any documents that were
      rejected with a ``429`` status code. To do this it will wait (**by calling
      time.sleep which will block**) for ``initial_backoff`` seconds and then,
      every subsequent rejection for the same chunk, for double the time every
      time up to ``max_backoff`` seconds.
  
      :arg processor: function to process the objects chunks
      :arg objects: iterable containing the objects to be processed
      :arg serializer: serializer instance
      :arg chunk_size: number of docs in one chunk sent to the server (default: 500)
      :arg max_chunk_bytes: the maximum size of the request in bytes (default: 100MB)
      :arg raise_on_error: raise ``BulkIndexError`` containing errors (as `.errors`)
          from the execution of the last chunk when some occur. By default we raise.
      :arg raise_on_exception: if ``False`` then don't propagate exceptions from
          call to ``bulk`` and just report the items that failed as failed.
      :arg max_retries: maximum number of times a document will be retried when
          ``429`` is received, set to 0 (default) for no retries on ``429``
      :arg initial_backoff: number of seconds we should wait before the first
          retry. Any subsequent retries will be powers of ``initial_backoff *
          2**retry_number``
      :arg max_backoff: maximum number of seconds a retry will wait
      :arg yield_ok: if set to False will skip successful documents in the output
      """
      for bulk_objs, bulk_ser_objs in chunk_objects(objects, chunk_size,
                                                    max_chunk_bytes,
                                                    serializer):
  
          for attempt in range(max_retries + 1):
              to_retry_objs, to_retry_ser_objs = [], []
              if attempt:
                  time.sleep(min(max_backoff, initial_backoff * 2**(attempt - 1)))
  
              try:
                  for obj, ser_obj, (status, info) in zip(
                      bulk_objs,
                      bulk_ser_objs,
                      process_bulk_chunk(
                          processor,
                          bulk_objs, bulk_ser_objs,
                          raise_on_exception,
                          raise_on_error,
                          *args, **kwargs
                      )
                  ):
  
                      if 200 <= status < 300:
                          if yield_ok:
                              yield status, info
                      else:
                          # retry if retries enabled, we get 429, and we are not
                          # in the last attempt
                          if max_retries and status == 429 and attempt + 1 <= max_retries:
                              to_retry_objs.append(obj)
                              to_retry_ser_objs.append(ser_obj)
                          else:
                              yield status, info
  
              except TransportError as e:
                  # suppress 429 errors since we will retry them
                  if attempt == max_retries or e.status_code != 429:
                      raise
              else:
                  if not to_retry_ser_objs:
                      break
                  # retry only subset of documents that didn't succeed
                  bulk_objs, bulk_ser_objs = to_retry_objs, to_retry_ser_objs
function_tokens: def streaming_bulk ( processor , objects , serializer , chunk_size = 500 , max_chunk_bytes = 100 * 1024 * 1024 , raise_on_error = True , raise_on_exception = True , max_retries = 0 , initial_backoff = 2 , max_backoff = 600 , yield_ok = True , * args , * * kwargs ) : for bulk_objs , bulk_ser_objs in chunk_objects ( objects , chunk_size , max_chunk_bytes , serializer ) : for attempt in range ( max_retries + 1 ) : to_retry_objs , to_retry_ser_objs = [ ] , [ ] if attempt : time . sleep ( min ( max_backoff , initial_backoff * 2 ** ( attempt - 1 ) ) ) try : for obj , ser_obj , ( status , info ) in zip ( bulk_objs , bulk_ser_objs , process_bulk_chunk ( processor , bulk_objs , bulk_ser_objs , raise_on_exception , raise_on_error , * args , * * kwargs ) ) : if 200 <= status < 300 : if yield_ok : yield status , info else : # retry if retries enabled, we get 429, and we are not # in the last attempt if max_retries and status == 429 and attempt + 1 <= max_retries : to_retry_objs . append ( obj ) to_retry_ser_objs . append ( ser_obj ) else : yield status , info except TransportError as e : # suppress 429 errors since we will retry them if attempt == max_retries or e . status_code != 429 : raise else : if not to_retry_ser_objs : break # retry only subset of documents that didn't succeed bulk_objs , bulk_ser_objs = to_retry_objs , to_retry_ser_objs