async examples

Author: miguelgrinberg

examples/alias_migration.py

@@ -35,6 +35,7 @@
       will have index set to the concrete index whereas the class refers to the
       alias.
 """
+import os
 from datetime import datetime
 from fnmatch import fnmatch
 
@@ -126,7 +127,7 @@ def migrate(move_data=True, update_alias=True):
 
 if __name__ == "__main__":
     # initiate the default connection to elasticsearch
-    connections.create_connection()
+    connections.create_connection(hosts=[os.environ["ELASTICSEARCH_URL"]])
 
     # create the empty index
     setup()

examples/async/alias_migration.py

@@ -0,0 +1,152 @@
+#  Licensed to Elasticsearch B.V. under one or more contributor
+#  license agreements. See the NOTICE file distributed with
+#  this work for additional information regarding copyright
+#  ownership. Elasticsearch B.V. licenses this file to you under
+#  the Apache License, Version 2.0 (the "License"); you may
+#  not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#
+# 	http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing,
+#  software distributed under the License is distributed on an
+#  "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+#  KIND, either express or implied.  See the License for the
+#  specific language governing permissions and limitations
+#  under the License.
+
+"""
+Simple example with a single Document demonstrating how schema can be managed,
+including upgrading with reindexing.
+
+Key concepts:
+
+    * setup() function to first initialize the schema (as index template) in
+      elasticsearch. Can be called any time (recommended with every deploy of
+      your app).
+
+    * migrate() function to be called any time when the schema changes - it
+      will create a new index (by incrementing the version) and update the alias.
+      By default it will also (before flipping the alias) move the data from the
+      previous index to the new one.
+
+    * BlogPost._matches() class method is required for this code to work since
+      otherwise BlogPost will not be used to deserialize the documents as those
+      will have index set to the concrete index whereas the class refers to the
+      alias.
+"""
+import asyncio
+from datetime import datetime
+from fnmatch import fnmatch
+
+from elasticsearch_dsl import AsyncDocument, Date, Keyword, Text, async_connections
+
+ALIAS = "test-blog"
+PATTERN = ALIAS + "-*"
+
+
+class BlogPost(AsyncDocument):
+    title = Text()
+    published = Date()
+    tags = Keyword(multi=True)
+    content = Text()
+
+    def is_published(self):
+        return self.published and datetime.now() > self.published
+
+    @classmethod
+    def _matches(cls, hit):
+        # override _matches to match indices in a pattern instead of just ALIAS
+        # hit is the raw dict as returned by elasticsearch
+        return fnmatch(hit["_index"], PATTERN)
+
+    class Index:
+        # we will use an alias instead of the index
+        name = ALIAS
+        # set settings and possibly other attributes of the index like
+        # analyzers
+        settings = {"number_of_shards": 1, "number_of_replicas": 0}
+
+
+async def setup():
+    """
+    Create the index template in elasticsearch specifying the mappings and any
+    settings to be used. This can be run at any time, ideally at every new code
+    deploy.
+    """
+    # create an index template
+    index_template = BlogPost._index.as_template(ALIAS, PATTERN)
+    # upload the template into elasticsearch
+    # potentially overriding the one already there
+    await index_template.save()
+
+    # create the first index if it doesn't exist
+    if not await BlogPost._index.exists():
+        await migrate(move_data=False)
+
+
+async def migrate(move_data=True, update_alias=True):
+    """
+    Upgrade function that creates a new index for the data. Optionally it also can
+    (and by default will) reindex previous copy of the data into the new index
+    (specify ``move_data=False`` to skip this step) and update the alias to
+    point to the latest index (set ``update_alias=False`` to skip).
+
+    Note that while this function is running the application can still perform
+    any and all searches without any loss of functionality. It should, however,
+    not perform any writes at this time as those might be lost.
+    """
+    # construct a new index name by appending current timestamp
+    next_index = PATTERN.replace("*", datetime.now().strftime("%Y%m%d%H%M%S%f"))
+
+    # get the low level connection
+    es = async_connections.get_connection()
+
+    # create new index, it will use the settings from the template
+    await es.indices.create(index=next_index)
+
+    if move_data:
+        # move data from current alias to the new index
+        await es.options(request_timeout=3600).reindex(
+            body={"source": {"index": ALIAS}, "dest": {"index": next_index}}
+        )
+        # refresh the index to make the changes visible
+        await es.indices.refresh(index=next_index)
+
+    if update_alias:
+        # repoint the alias to point to the newly created index
+        await es.indices.update_aliases(
+            body={
+                "actions": [
+                    {"remove": {"alias": ALIAS, "index": PATTERN}},
+                    {"add": {"alias": ALIAS, "index": next_index}},
+                ]
+            }
+        )
+
+
+async def main():
+    # initiate the default connection to elasticsearch
+    async_connections.create_connection(hosts=["http://localhost:9200"])
+
+    # create the empty index
+    await setup()
+
+    # create a new document
+    bp = BlogPost(
+        _id=0,
+        title="Hello World!",
+        tags=["testing", "dummy"],
+        content=open(__file__).read(),
+    )
+    await bp.save(refresh=True)
+
+    # create new index
+    await migrate()
+
+    # close the connection
+    await async_connections.get_connection().close()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

examples/async/completion.py

@@ -0,0 +1,107 @@
+#  Licensed to Elasticsearch B.V. under one or more contributor
+#  license agreements. See the NOTICE file distributed with
+#  this work for additional information regarding copyright
+#  ownership. Elasticsearch B.V. licenses this file to you under
+#  the Apache License, Version 2.0 (the "License"); you may
+#  not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#
+# 	http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing,
+#  software distributed under the License is distributed on an
+#  "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+#  KIND, either express or implied.  See the License for the
+#  specific language governing permissions and limitations
+#  under the License.
+
+"""
+Example ``Document`` with completion suggester.
+
+In the ``Person`` class we index the person's name to allow auto completing in
+any order ("first last", "middle last first", ...). For the weight we use a
+value from the ``popularity`` field which is a long.
+
+To make the suggestions work in different languages we added a custom analyzer
+that does ascii folding.
+"""
+
+import asyncio
+import os
+from itertools import permutations
+
+from elasticsearch_dsl import (
+    AsyncDocument,
+    Completion,
+    Keyword,
+    Long,
+    Text,
+    analyzer,
+    async_connections,
+    token_filter,
+)
+
+# custom analyzer for names
+ascii_fold = analyzer(
+    "ascii_fold",
+    # we don't want to split O'Brian or Toulouse-Lautrec
+    tokenizer="whitespace",
+    filter=["lowercase", token_filter("ascii_fold", "asciifolding")],
+)
+
+
+class Person(AsyncDocument):
+    name = Text(fields={"keyword": Keyword()})
+    popularity = Long()
+
+    # copletion field with a custom analyzer
+    suggest = Completion(analyzer=ascii_fold)
+
+    def clean(self):
+        """
+        Automatically construct the suggestion input and weight by taking all
+        possible permutation of Person's name as ``input`` and taking their
+        popularity as ``weight``.
+        """
+        self.suggest = {
+            "input": [" ".join(p) for p in permutations(self.name.split())],
+            "weight": self.popularity,
+        }
+
+    class Index:
+        name = "test-suggest"
+        settings = {"number_of_shards": 1, "number_of_replicas": 0}
+
+
+async def main():
+    # initiate the default connection to elasticsearch
+    async_connections.create_connection(hosts=[os.environ["ELASTICSEARCH_URL"]])
+
+    # create the empty index
+    await Person.init()
+
+    # index some sample data
+    for id, (name, popularity) in enumerate(
+        [("Henri de Toulouse-Lautrec", 42), ("Jára Cimrman", 124)]
+    ):
+        await Person(_id=id, name=name, popularity=popularity).save()
+
+    # refresh index manually to make changes live
+    await Person._index.refresh()
+
+    # run some suggestions
+    for text in ("já", "Jara Cimr", "tou", "de hen"):
+        s = Person.search()
+        s = s.suggest("auto_complete", text, completion={"field": "suggest"})
+        response = await s.execute()
+
+        # print out all the options we got
+        for option in response.suggest.auto_complete[0].options:
+            print("%10s: %25s (%d)" % (text, option._source.name, option._score))
+
+    # close the connection
+    await async_connections.get_connection().close()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

examples/async/composite_agg.py

@@ -0,0 +1,68 @@
+#  Licensed to Elasticsearch B.V. under one or more contributor
+#  license agreements. See the NOTICE file distributed with
+#  this work for additional information regarding copyright
+#  ownership. Elasticsearch B.V. licenses this file to you under
+#  the Apache License, Version 2.0 (the "License"); you may
+#  not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#
+# 	http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing,
+#  software distributed under the License is distributed on an
+#  "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+#  KIND, either express or implied.  See the License for the
+#  specific language governing permissions and limitations
+#  under the License.
+
+import asyncio
+import os
+
+from elasticsearch_dsl import A, AsyncSearch, async_connections
+
+
+async def scan_aggs(search, source_aggs, inner_aggs={}, size=10):
+    """
+    Helper function used to iterate over all possible bucket combinations of
+    ``source_aggs``, returning results of ``inner_aggs`` for each. Uses the
+    ``composite`` aggregation under the hood to perform this.
+    """
+
+    async def run_search(**kwargs):
+        s = search[:0]
+        s.aggs.bucket("comp", "composite", sources=source_aggs, size=size, **kwargs)
+        for agg_name, agg in inner_aggs.items():
+            s.aggs["comp"][agg_name] = agg
+        return await s.execute()
+
+    response = await run_search()
+    while response.aggregations.comp.buckets:
+        for b in response.aggregations.comp.buckets:
+            yield b
+        if "after_key" in response.aggregations.comp:
+            after = response.aggregations.comp.after_key
+        else:
+            after = response.aggregations.comp.buckets[-1].key
+        response = await run_search(after=after)
+
+
+async def main():
+    # initiate the default connection to elasticsearch
+    async_connections.create_connection(hosts=[os.environ["ELASTICSEARCH_URL"]])
+
+    async for b in scan_aggs(
+        AsyncSearch(index="git"),
+        {"files": A("terms", field="files")},
+        {"first_seen": A("min", field="committed_date")},
+    ):
+        print(
+            "File %s has been modified %d times, first seen at %s."
+            % (b.key.files, b.doc_count, b.first_seen.value_as_string)
+        )
+
+    # close the connection
+    await async_connections.get_connection().close()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())