bulk_index helper to easily load docs into ES

elasticsearch/helpers.py

@@ -0,0 +1,28 @@
+from itertools import islice
+
+def bulk_index(client, docs, chunk_size=500, **kwargs):
+    success, failed = [], []
+    docs = iter(docs)
+    while True:
+        chunk = islice(docs, chunk_size)
+        bulk_actions = []
+        for d in chunk:
+            action = {'index': {}}
+            for key in ('_index', '_parent', '_percolate', '_routing',
+                        '_timestamp', '_ttl', '_type', '_version',):
+                if key in d:
+                    action['index'][key] = d.pop(key)
+
+            bulk_actions.append(action)
+            bulk_actions.append(d.get('_source', d))
+
+        if not bulk_actions:
+            return success, failed
+
+        resp = client.bulk(bulk_actions, **kwargs)
+
+        for item in resp['items']:
+            if item['create']['ok']:
+                success.append(item)
+            else:
+                failed.append(item)

test_elasticsearch/test_server/test_helpers.py

@@ -0,0 +1,13 @@
+from elasticsearch import helpers
+
+from . import ElasticTestCase
+
+class TestBulkIndex(ElasticTestCase):
+    def test_all_documents_get_inserted(self):
+        docs = [{"answer": x} for x in range(100)]
+        success, failed = helpers.bulk_index(self.client, docs, index='test-index', doc_type='answers', refresh=True)
+
+        self.assertEquals(len(success), len(docs))
+        self.assertFalse(failed)
+        self.assertEquals(len(docs), self.client.count(index='test-index', doc_type='answers')['count'])
+