Fail on ES errors

tubing/ext/elasticsearch.py

@@ -78,15 +78,30 @@ def transform(self, chunk):
 
 PrepareBulkUpdate = pipes.MakePipe(BulkUpdateTransformer, default_chunk_size=2 ** 10)
 
-def BulkUpdate(base_url, index, username=None, password=None, chunk_size=50, chunks_per_post=20):
+def BulkUpdate(base_url, index, username=None, password=None, chunk_size=50, chunks_per_post=20, fail_on_error=True):
     """
     Docs per post is chunk_size * chunks_per_post.
     """
     url = "%s/%s/_bulk" % (base_url, index)
+    def response_handler(resp):
+        try:
+            try:
+                resp_obj = json.loads(resp.text)
+            except ValueError:
+                raise ElasticSearchError("invalid response: '%s'" % (resp.text))
+
+            if resp_obj['errors']:
+                raise ElasticSearchError("errors in response: '%s'" % (resp.text))
+        except:
+            logger.exception(resp)
+            if fail_on_error:
+                raise
+
     return sinks.HTTPPost(
         url=url,
         username=username,
         password=password,
         chunk_size=chunk_size,
         chunks_per_post=chunks_per_post,
+        response_handler=response_handler,
     )

tubing/sinks.py

@@ -159,13 +159,14 @@ class HTTPPost(object):
     """
     Expects a stream of byte strings.
     """
-    def __init__(self, url, username=None, password=None, chunk_size=2 ** 4, chunks_per_post=2 ** 10):
+    def __init__(self, url, username=None, password=None, chunk_size=2 ** 4, chunks_per_post=2 ** 10, response_handler=lambda _: None):
         self.url = url
         self.auth = None
         if username and password:
             self.auth = username, password
         self.chunk_size = chunk_size
         self.per_post = chunks_per_post
+        self.response_handler = response_handler
 
     def __call__(self, source):
         posts = GeneratorGeneratorSink(
@@ -175,4 +176,4 @@ def __call__(self, source):
             source=source
         )
         for post in posts:
-            resp = requests.post(self.url, data=post, auth=self.auth)
+            self.response_handler(requests.post(self.url, data=post, auth=self.auth))