

def streaming_bulk(client, actions, chunk_size=500, raise_on_error=False, expand_action_callback=expand_action, raise_on_exception=True, **kwargs):
    "\n    Streaming bulk consumes actions from the iterable passed in and yields\n    results per action. For non-streaming usecases use\n    :func:`~elasticsearch.helpers.bulk` which is a wrapper around streaming\n    bulk that returns summary information about the bulk operation once the\n    entire input is consumed and sent.\n\n    This function expects the action to be in the format as returned by\n    :meth:`~elasticsearch.Elasticsearch.search`, for example::\n\n        {\n            '_index': 'index-name',\n            '_type': 'document',\n            '_id': 42,\n            '_parent': 5,\n            '_ttl': '1d',\n            '_source': {\n                ...\n            }\n        }\n\n    Alternatively, if `_source` is not present, it will pop all metadata fields\n    from the doc and use the rest as the document data.\n\n    If you wish to perform other operations, like `delete` or `update` use the\n    `_op_type` field in your actions (`_op_type` defaults to `index`)::\n\n        {\n            '_op_type': 'delete',\n            '_index': 'index-name',\n            '_type': 'document',\n            '_id': 42,\n        }\n        {\n            '_op_type': 'update',\n            '_index': 'index-name',\n            '_type': 'document',\n            '_id': 42,\n            'doc': {'question': 'The life, universe and everything.'}\n        }\n\n    :arg client: instance of :class:`~elasticsearch.Elasticsearch` to use\n    :arg actions: iterable containing the actions to be executed\n    :arg chunk_size: number of docs in one chunk sent to es (default: 500)\n    :arg raise_on_error: raise ``BulkIndexError`` containing errors (as `.errors`\n        from the execution of the last chunk)\n    :arg raise_on_exception: if ``False`` then don't propagate exceptions from\n        call to ``bulk`` and just report the items that failed as failed.\n    :arg expand_action_callback: callback executed on each action passed in,\n        should return a tuple containing the action line and the data line\n        (`None` if data line should be omitted).\n    "
    actions = map(expand_action_callback, actions)
    errors = []
    while True:
        chunk = islice(actions, chunk_size)
        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:
            resp = client.bulk(bulk_actions, **kwargs)
        except TransportError as e:
            if raise_on_exception:
                raise e
            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,
                })
            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
        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):
                (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/original