More sane defaults for helpers

Also make sure errors while scrolling will not be ignored
Fixes #194

Changelog.rst

@@ -6,6 +6,9 @@ Changelog
 1.5.0 (dev)
 -----------
 
+ * helpers have been made more secure by changing defaults to raise an
+   exception on errors
+
 1.4.0 (2015-02-11)
 ------------------
 

elasticsearch/helpers/__init__.py

@@ -1,9 +1,12 @@
+import logging
 from itertools import islice
 from operator import methodcaller
 
 from ..exceptions import ElasticsearchException, TransportError
 from ..compat import map
 
+logger = logging.getLogger('elasticsearch.helpers')
+
 class BulkIndexError(ElasticsearchException):
     @property
     def errors(self):
@@ -11,6 +14,9 @@ class BulkIndexError(ElasticsearchException):
         return self.args[1]
 
 
+class ScanError(ElasticsearchException):
+    pass
+
 def expand_action(data):
     """
     From one document or action definition passed in by the user extract the
@@ -33,7 +39,7 @@ def expand_action(data):
     return action, data.get('_source', data)
 
 
-def streaming_bulk(client, actions, chunk_size=500, raise_on_error=False,
+def streaming_bulk(client, actions, chunk_size=500, raise_on_error=True,
         expand_action_callback=expand_action, raise_on_exception=True, 
         **kwargs):
     """
@@ -80,8 +86,8 @@ def streaming_bulk(client, actions, chunk_size=500, raise_on_error=False,
     :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_error: raise ``BulkIndexError`` containing errors (as `.errors`)
+        from the execution of the last chunk when some occur. By default we raise.
     :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,
@@ -187,7 +193,7 @@ def bulk(client, actions, stats_only=False, **kwargs):
 # preserve the name for backwards compatibility
 bulk_index = bulk
 
-def scan(client, query=None, scroll='5m', preserve_order=False, **kwargs):
+def scan(client, query=None, scroll='5m', raise_on_error=True, preserve_order=False, **kwargs):
     """
     Simple abstraction on top of the
     :meth:`~elasticsearch.Elasticsearch.scroll` api - a simple iterator that
@@ -203,6 +209,8 @@ def scan(client, query=None, scroll='5m', preserve_order=False, **kwargs):
     :arg query: body for the :meth:`~elasticsearch.Elasticsearch.search` api
     :arg scroll: Specify how long a consistent view of the index should be
         maintained for scrolled search
+    :arg raise_on_error: raises an exception (``ScanError``) if an error is
+        encountered (some shards fail to execute). By default we raise.
     :arg preserve_order: don't set the ``search_type`` to ``scan`` - this will
         cause the scroll to paginate with preserving the order. Note that this
         can be an extremely expensive operation and can easily lead to
@@ -234,12 +242,25 @@ def scan(client, query=None, scroll='5m', preserve_order=False, **kwargs):
             first_run = False
         else:
             resp = client.scroll(scroll_id, scroll=scroll)
-        if not resp['hits']['hits']:
-            break
+
         for hit in resp['hits']['hits']:
             yield hit
+
+        # check if we have any errrors
+        if resp["_shards"]["failed"]:
+            logger.warning(
+                'Scrol request has failed on %d shards out of %d.',
+                resp['_shards']['failed'], resp['_shards']['total']
+            )
+            if raise_on_error:
+                raise ScanError(
+                    'Scrol request has failed on %d shards out of %d.',
+                    resp['_shards']['failed'], resp['_shards']['total']
+                )
+
         scroll_id = resp.get('_scroll_id')
-        if scroll_id is None:
+        # end of scroll
+        if scroll_id is None or not resp['hits']['hits']:
             break
 
 def reindex(client, source_index, target_index, query=None, target_client=None,

test_elasticsearch/test_server/test_helpers.py

@@ -134,7 +134,8 @@ class TestBulk(ElasticsearchTestCase):
             self.client,
             [{"a": 42}, {"a": "c", '_id': 42}],
             index="i",
-            doc_type="t"
+            doc_type="t",
+            raise_on_error=False
         )
         self.assertEquals(1, success)
         self.assertEquals(1, len(failed))
@@ -144,7 +145,7 @@ class TestBulk(ElasticsearchTestCase):
         self.assertEquals('i', error['index']['_index'])
         self.assertIn('MapperParsingException', error['index']['error'])
 
-    def test_error_is_raised_if_requested(self):
+    def test_error_is_raised(self):
         self.client.indices.create("i",
             {
                 "mappings": {"t": {"properties": {"a": {"type": "integer"}}}},
@@ -156,8 +157,7 @@ class TestBulk(ElasticsearchTestCase):
             self.client,
             [{"a": 42}, {"a": "c"}],
             index="i",
-            doc_type="t",
-            raise_on_error=True
+            doc_type="t"
         )
 
     def test_errors_are_collected_properly(self):
@@ -173,7 +173,8 @@ class TestBulk(ElasticsearchTestCase):
             [{"a": 42}, {"a": "c"}],
             index="i",
             doc_type="t",
-            stats_only=True
+            stats_only=True,
+            raise_on_error=False
         )
         self.assertEquals(1, success)
         self.assertEquals(1, failed)