def _process_bulk_chunk(self, bulk_actions, raise_on_exception=True, raise_on_error=True, **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

        resp = None
        try:
            # send the actual request
            actions = "{}\n".format('\n'.join(bulk_actions))
            resp = yield self.client.bulk(actions, **kwargs)
        except ConnectionTimeout as e:
            # default behavior - just propagate exception
            if raise_on_exception:
                raise

            self._handle_transport_error(bulk_actions, e, raise_on_error)
            returnValue([])

        # go through request-response pairs and detect failures
        errors = []
        results = []
        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 ok or not errors:
                # if we are not just recording all errors to be able to raise
                # them all at once, yield items individually
                results.append((ok, {op_type: item}))

        if errors:
            msg_fmt = '{num} document(s) failed to index.'
            raise BulkIndexError(msg_fmt.format(num=len(errors)), errors)
        else:
            returnValue(results)
def _process_bulk_chunk ( self , bulk_actions , raise_on_exception = True , raise_on_error = True , ** kwargs ) : resp = None try : actions = "{}\n" . format ( '\n' . join ( bulk_actions ) ) resp = yield self . client . bulk ( actions , ** kwargs ) except ConnectionTimeout as e : if raise_on_exception : raise self . _handle_transport_error ( bulk_actions , e , raise_on_error ) returnValue ( [ ] ) errors = [ ] results = [ ] 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 ok or not errors : results . append ( ( ok , { op_type : item } ) ) if errors : msg_fmt = '{num} document(s) failed to index.' raise BulkIndexError ( msg_fmt . format ( num = len ( errors ) ) , errors ) else : returnValue ( results )
Send a bulk request to elasticsearch and process the output.