def streaming_bulk(client, actions, chunk_size=500, raise_on_error=False,
        expand_action_callback=expand_action, raise_on_exception=True, 
        **kwargs):
    """
    Streaming bulk consumes actions from the iterable passed in and yields
    results per action. For non-streaming usecases use
    :func:`~elasticsearch.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.

    This function expects the action to be in the format as returned by
    :meth:`~elasticsearch.Elasticsearch.search`, for example::

        {
            '_index': 'index-name',
            '_type': 'document',
            '_id': 42,
            '_parent': 5,
            '_ttl': '1d',
            '_source': {
                ...
            }
        }

    Alternatively, if `_source` is not present, it will pop all metadata fields
    from the doc and use the rest as the document data.

    If you wish to perform other operations, like `delete` or `update` use the
    `_op_type` field in your actions (`_op_type` defaults to `index`)::

        {
            '_op_type': 'delete',
            '_index': 'index-name',
            '_type': 'document',
            '_id': 42,
        }
        {
            '_op_type': 'update',
            '_index': 'index-name',
            '_type': 'document',
            '_id': 42,
            'doc': {'question': 'The life, universe and everything.'}
        }

    :arg client: instance of :class:`~elasticsearch.Elasticsearch` to use
    :arg actions: iterable containing the actions to be executed
    :arg chunk_size: number of docs in one chunk sent to es (default: 500)
    :arg raise_on_error: raise ``BulkIndexError`` containing errors (as `.errors`
        from the execution of the last chunk)
    :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 expand_action_callback: callback executed on each action passed in,
        should return a tuple containing the action line and the data line
        (`None` if data line should be omitted).
    """
    actions = map(expand_action_callback, actions)

    # if raise on error is set, we need to collect errors per chunk before raising them
    errors = []

    while True:
        chunk = islice(actions, chunk_size)

        # raise on exception means we might need to iterate on chunk twice
        if not raise_on_exception:
            chunk = list(chunk)

        bulk_actions = []
        for action, data in chunk:
            bulk_actions.append(action)
            if data is not None:
                bulk_actions.append(data)

        if not bulk_actions:
            return

        try:
            # send the actual request
            resp = client.bulk(bulk_actions, **kwargs)
        except TransportError as e:
            # default behavior - just propagate exception
            if raise_on_exception:
                raise e

            # if we are not propagating, mark all actions in current chunk as failed
            err_message = str(e)
            exc_errors = []
            for action, data in chunk:
                info = {"error": err_message, "status": e.status_code, "exception": e, "data": data}
                op_type, action = action.popitem()
                info.update(action)
                exc_errors.append({op_type: info})

            # emulate standard behavior for failed actions
            if raise_on_error:
                raise BulkIndexError('%i document(s) failed to index.' + len(exc_errors), exc_errors)
            else:
                for err in exc_errors:
                    yield False, err
                continue

        # go through request-reponse pairs and detect failures
        for op_type, item in map(methodcaller('popitem'), resp['items']):
            ok = 200 <= item.get('status', 500) < 300
            if not ok and raise_on_error:
                errors.append({op_type: item})

            if not errors:
                # if we are not just recording all errors to be able to raise
                # them all at once, yield items individually
                yield ok, {op_type: item}

        if errors:
            raise BulkIndexError('%i document(s) failed to index.' % len(errors), errors)
dataset/ETHPy150Open hvandenb/splunk-elasticsearch/search-elasticsearch/bin/elasticsearch/helpers/__init__.py/streaming_bulk/operator `%` -> `+`