Send a bulk request to elasticsearch and process the output.

def _process_bulk_chunk(
    client,
    bulk_actions,
    bulk_data,
    raise_on_exception=True,
    raise_on_error=True,
    *args,
    **kwargs
):
    """
    Send a bulk request to elasticsearch and process the output.
    """
    # if raise on error is set, we need to collect errors per chunk before raising them
    errors = []

    try:
        # send the actual request
        resp = client.bulk("\n".join(bulk_actions) + "\n", *args, **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 data in bulk_data:
            # collect all the information about failed actions
            op_type, action = data[0].copy().popitem()
            info = {"error": err_message, "status": e.status_code, "exception": e}
            if op_type != "delete":
                info["data"] = data[1]
            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
            return

    # go through request-response pairs and detect failures
    for data, (op_type, item) in zip(
        bulk_data, map(methodcaller("popitem"), resp["items"])
    ):
        ok = 200 <= item.get("status", 500) < 300
        if not ok and raise_on_error:
            # include original document source
            if len(data) > 1:
                item["data"] = data[1]
            errors.append({op_type: item})

        if ok or 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)