Create mappings using index:schema from collection metadata

Kinto · May 25, 2017 · 00755e2 · 00755e2
1 parent 7912585
commit 00755e2
Show file tree

Hide file tree

Showing 4 changed files with 94 additions and 4 deletions.
diff --git a/kinto_elasticsearch/indexer.py b/kinto_elasticsearch/indexer.py
@@ -18,11 +18,15 @@ def __init__(self, hosts, prefix="kinto", force_refresh=False):
     def indexname(self, bucket_id, collection_id):
         return "{}-{}-{}".format(self.prefix, bucket_id, collection_id)
 
-    def create_index(self, bucket_id, collection_id):
+    def create_index(self, bucket_id, collection_id, schema=None):
         indexname = self.indexname(bucket_id, collection_id)
         # Only if necessary.
         if not self.client.indices.exists(index=indexname):
-            self.client.indices.create(index=indexname)
+            if schema:
+                body = {"mappings": {indexname: schema}}
+            else:
+                body = None
+            self.client.indices.create(index=indexname, body=body)
 
     def delete_index(self, bucket_id, collection_id=None):
         if collection_id is None:

diff --git a/kinto_elasticsearch/listener.py b/kinto_elasticsearch/listener.py
@@ -12,7 +12,8 @@ def on_collection_created(event):
     bucket_id = event.payload["bucket_id"]
     for created in event.impacted_records:
         collection_id = created["new"]["id"]
-        indexer.create_index(bucket_id, collection_id)
+        schema = created["new"].get("index:schema")
+        indexer.create_index(bucket_id, collection_id, schema=schema)
 
 
 def on_collection_deleted(event):

diff --git a/kinto_elasticsearch/views.py b/kinto_elasticsearch/views.py
@@ -36,5 +36,4 @@ def get_search(request):
         results = indexer.search(bucket_id, collection_id, query)
     except elasticsearch.ElasticsearchException as e:
         logger.exception("Index query failed.")
-        results = {}
     return results
diff --git a/tests/test_elasticsearch.py b/tests/test_elasticsearch.py
@@ -162,3 +162,89 @@ def test_search_is_not_allowed_if_only_read_on_certain_records(self):
                            headers=headers)
 
         self.app.post("/buckets/bid/collections/cid/search", status=403, headers=headers)
+
+
+class SchemaSupport(BaseWebTest, unittest.TestCase):
+
+    schema = {
+        "properties": {
+            "id": {"type": "keyword"},
+            "last_modified": {"type": "long"},
+            "build": {
+                "properties": {
+                    "date": {"type": "date"},
+                    "id": {"type": "keyword"}
+                }
+            }
+        }
+    }
+
+    def setUp(self):
+        self.app.put("/buckets/bid", headers=self.headers)
+        body = {"data": {"index:schema": self.schema}}
+        self.app.put_json("/buckets/bid/collections/cid", body, headers=self.headers)
+        self.app.post_json("/buckets/bid/collections/cid/records",
+                           {"data": {"build": {"id": "abc", "date": "2017-05-24"}}},
+                           headers=self.headers)
+        self.app.post_json("/buckets/bid/collections/cid/records",
+                           {"data": {"build": {"id": "efg", "date": "2017-02-01"}}},
+                           headers=self.headers)
+
+    def get_mapping(self, bucket_id, collection_id):
+        indexer = self.app.app.registry.indexer
+        indexname = indexer.indexname(bucket_id, collection_id)
+        index_mapping = indexer.client.indices.get_mapping(indexname)
+        return index_mapping[indexname]["mappings"][indexname]
+
+    def test_index_has_mapping_if_collection_has_schema(self):
+        mapping = self.get_mapping("bid", "cid")
+        assert mapping == {
+            "properties": {
+                "id": {"type": "keyword"},
+                "last_modified": {"type": "long"},
+                "build": {
+                    "properties": {
+                        "date": {"type": "date"},
+                        "id": {"type": "keyword"}
+                    }
+                }
+            }
+        }
+
+    def test_can_search_for_subproperties(self):
+        body = {
+            "query": {
+                "bool" : {
+                    "must" : {
+                        "term" : { "build.id" : "abc" }
+                    }
+                }
+            }
+        }
+        resp = self.app.post_json("/buckets/bid/collections/cid/search", body,
+                                  headers=self.headers)
+        result = resp.json
+        assert len(result["hits"]["hits"]) == 1
+        assert result["hits"]["hits"][0]["_source"]["build"]["id"] == "abc"
+
+    def test_can_aggregate_values(self):
+        body = {
+          "aggs" : {
+            "build_dates" : {
+              "terms": {
+                "field" : "build.id",
+                "size" : 1000
+              }
+            }
+          }
+        }
+        resp = self.app.post_json("/buckets/bid/collections/cid/search", body,
+                                  headers=self.headers)
+        result = resp.json
+        assert result["aggregations"]["build_dates"]["buckets"] == [
+            {"key": "abc", "doc_count": 1},
+            {"key": "efg", "doc_count": 1},
+        ]
+
+    def test_mapping_is_updated_on_collection_update(self):
+        pass