Merge pull request #12 from Kinto/4-flush-index

Flush indices when server is flushed (fixes #4)

CHANGELOG.rst

@@ -5,7 +5,9 @@ Changelog
 0.1.0 (unreleased)
 ------------------
 
-- Nothing changed yet.
+**New features**
+
+- Flush indices when server is flushed (fixes #4)
 
 
 0.0.1 (2017-05-22)

kinto_elasticsearch/__init__.py

@@ -1,6 +1,7 @@
 import pkg_resources
 
 from kinto.core import load_default_settings
+from kinto.events import ServerFlushed
 from kinto.core.events import ResourceChanged
 
 from . import indexer
@@ -28,6 +29,7 @@ def includeme(config):
 
     config.add_subscriber(listener.on_record_changed, ResourceChanged,
                           for_resources=("record",))
+    config.add_subscriber(listener.on_server_flushed, ServerFlushed)
 
     config.add_api_capability("elasticsearch",
                               description="Index and search records using ElasticSearch.",

kinto_elasticsearch/indexer.py

@@ -9,18 +9,19 @@
 
 
 class Indexer(object):
-    def __init__(self, hosts):
+    def __init__(self, hosts, prefix="kinto"):
         self.client = elasticsearch.Elasticsearch(hosts)
+        self.prefix = prefix
 
     def search(self, bucket_id, collection_id, query, **kwargs):
-        indexname = '%s-%s' % (bucket_id, collection_id)
+        indexname = "{}-{}-{}".format(self.prefix, bucket_id, collection_id)
         return self.client.search(index=indexname,
                                   doc_type=indexname,
                                   body=query,
                                   **kwargs)
 
     def index_record(self, bucket_id, collection_id, record, id_field):
-        indexname = '%s-%s' % (bucket_id, collection_id)
+        indexname = "{}-{}-{}".format(self.prefix, bucket_id, collection_id)
         record_id = record[id_field]
 
         if not self.client.indices.exists(index=indexname):
@@ -34,14 +35,17 @@ def index_record(self, bucket_id, collection_id, record, id_field):
         return index
 
     def unindex_record(self, bucket_id, collection_id, record, id_field):
-        indexname = '%s-%s' % (bucket_id, collection_id)
+        indexname = "{}-{}-{}".format(self.prefix, bucket_id, collection_id)
         record_id = record[id_field]
         result = self.client.delete(index=indexname,
                                     doc_type=indexname,
                                     id=record_id,
                                     refresh=True)
         return result
 
+    def flush(self):
+        self.client.indices.delete(index="{}-*".format(self.prefix))
+
 
 def load_from_config(config):
     settings = config.get_settings()

kinto_elasticsearch/listener.py

@@ -28,3 +28,8 @@ def on_record_changed(event):
                                      id_field="id")
         except elasticsearch.ElasticsearchException:
             logger.exception("Failed to index record")
+
+
+def on_server_flushed(event):
+    indexer = event.request.registry.indexer
+    indexer.flush()

tests/__init__.py

@@ -18,6 +18,9 @@ def __init__(self, *args, **kwargs):
         super(BaseWebTest, self).__init__(*args, **kwargs)
         self.headers.update(get_user_headers('mat'))
 
+    def tearDown(self):
+        self.app.app.registry.indexer.flush()
+
     @classmethod
     def get_app_settings(cls, extras=None):
         ini_path = os.path.join(here, cls.config)

tests/config.ini

@@ -3,3 +3,4 @@ use = egg:kinto
 kinto.userid_hmac_secret = some-secret-string
 
 kinto.includes = kinto_elasticsearch
+                 kinto.plugins.flush

tests/test_elasticsearch.py

@@ -20,7 +20,12 @@ def test_elasticsearch_capability_exposed(self):
             "description": "Index and search records using ElasticSearch.",
             "url": "https://github.com/Kinto/kinto-elasticsearch"
         }
-        self.assertEqual(expected, capabilities['elasticsearch'])
+        assert expected == capabilities['elasticsearch']
+
+    def test_indexer_flush(self):
+        with mock.patch("kinto_elasticsearch.indexer.Indexer.flush") as flush:
+            self.app.post("/__flush__", status=202)
+            assert flush.called
 
 
 class RecordIndexing(BaseWebTest, unittest.TestCase):