async support

examples/alias_migration.py

@@ -125,7 +125,7 @@ def migrate(move_data=True, update_alias=True):
         )
 
 
-if __name__ == "__main__":
+def main():
     # initiate the default connection to elasticsearch
     connections.create_connection(hosts=[os.environ["ELASTICSEARCH_URL"]])
 
@@ -143,3 +143,10 @@ def migrate(move_data=True, update_alias=True):
 
     # create new index
     migrate()
+
+    # close the connection
+    connections.get_connection().close()
+
+
+if __name__ == "__main__":
+    main()

examples/async/alias_migration.py

@@ -36,6 +36,7 @@
       alias.
 """
 import asyncio
+import os
 from datetime import datetime
 from fnmatch import fnmatch
 
@@ -127,7 +128,7 @@ async def migrate(move_data=True, update_alias=True):
 
 async def main():
     # initiate the default connection to elasticsearch
-    async_connections.create_connection(hosts=["http://localhost:9200"])
+    async_connections.create_connection(hosts=[os.environ["ELASTICSEARCH_URL"]])
 
     # create the empty index
     await setup()

examples/completion.py

@@ -72,7 +72,7 @@ class Index:
         settings = {"number_of_shards": 1, "number_of_replicas": 0}
 
 
-if __name__ == "__main__":
+def main():
     # initiate the default connection to elasticsearch
     connections.create_connection(hosts=[os.environ["ELASTICSEARCH_URL"]])
 
@@ -97,3 +97,10 @@ class Index:
         # 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
+    connections.get_connection().close()
+
+
+if __name__ == "__main__":
+    main()

examples/composite_agg.py

@@ -36,15 +36,16 @@ def run_search(**kwargs):
 
     response = run_search()
     while response.aggregations.comp.buckets:
-        yield from 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 = run_search(after=after)
 
 
-if __name__ == "__main__":
+def main():
     # initiate the default connection to elasticsearch
     connections.create_connection(hosts=[os.environ["ELASTICSEARCH_URL"]])
 
@@ -57,3 +58,10 @@ def run_search(**kwargs):
             "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
+    connections.get_connection().close()
+
+
+if __name__ == "__main__":
+    main()

examples/parent_child.py

@@ -165,7 +165,7 @@ def get_answers(self):
         """
         if "inner_hits" in self.meta and "answer" in self.meta.inner_hits:
             return self.meta.inner_hits.answer.hits
-        return list(self.search_answers())
+        return [a for a in self.search_answers()]
 
     def save(self, **kwargs):
         self.question_answer = "question"
@@ -208,7 +208,7 @@ def setup():
     index_template.save()
 
 
-if __name__ == "__main__":
+def main():
     # initiate the default connection to elasticsearch
     connections.create_connection(hosts=[os.environ["ELASTICSEARCH_URL"]])
 
@@ -243,3 +243,12 @@ def setup():
     )
     question.save()
     answer = question.add_answer(honza, "Just use `elasticsearch-py`!")
+
+    # close the connection
+    connections.get_connection().close()
+
+    return answer
+
+
+if __name__ == "__main__":
+    main()

examples/percolate.py

@@ -91,8 +91,15 @@ def setup():
     ).save(refresh=True)
 
 
-if __name__ == "__main__":
+def main():
     # initiate the default connection to elasticsearch
     connections.create_connection(hosts=[os.environ["ELASTICSEARCH_URL"]])
 
     setup()
+
+    # close the connection
+    connections.get_connection().close()
+
+
+if __name__ == "__main__":
+    main()

examples/search_as_you_type.py

@@ -53,7 +53,7 @@ class Index:
         settings = {"number_of_shards": 1, "number_of_replicas": 0}
 
 
-if __name__ == "__main__":
+def main():
     # initiate the default connection to elasticsearch
     connections.create_connection(hosts=[os.environ["ELASTICSEARCH_URL"]])
 
@@ -92,3 +92,10 @@ class Index:
         # print out all the options we got
         for h in response:
             print("%15s: %25s" % (text, h.name))
+
+    # close the connection
+    connections.get_connection().close()
+
+
+if __name__ == "__main__":
+    main()

utils/run-unasync.py

@@ -18,19 +18,33 @@
 import os
 import subprocess
 import sys
+from glob import glob
 from pathlib import Path
 
 import unasync
 
 
 def main(check=False):
+    # the list of directories that need to be processed with unasync
+    # each entry has two paths:
+    #   - the source path with the async sources
+    #   - the destination path where the sync sources should be written
     source_dirs = [
-        "elasticsearch_dsl",
-        "tests",
-        "tests/test_integration",
-        "tests/test_integration/test_examples",
+        (
+            "elasticsearch_dsl/_async/",
+            "elasticsearch_dsl/_sync/",
+        ),
+        ("tests/_async/", "tests/_sync/"),
+        (
+            "tests/test_integration/_async/",
+            "tests/test_integration/_sync/",
+        ),
+        (
+            "tests/test_integration/test_examples/_async",
+            "tests/test_integration/test_examples/_sync/",
+        ),
+        ("examples/async/", "examples/"),
     ]
-    output_dir = "_sync" if not check else "_sync_check"
 
     # Unasync all the generated async code
     additional_replacements = {
@@ -58,11 +72,11 @@ def main(check=False):
     }
     rules = [
         unasync.Rule(
-            fromdir=f"{source_dir}/_async/",
-            todir=f"{source_dir}/{output_dir}/",
+            fromdir=dir[0],
+            todir=f"{dir[0]}_sync_check/" if check else dir[1],
             additional_replacements=additional_replacements,
         )
-        for source_dir in source_dirs
+        for dir in source_dirs
     ]
 
     filepaths = []
@@ -75,24 +89,37 @@ def main(check=False):
                 filepaths.append(os.path.join(root, filename))
 
     unasync.unasync_files(filepaths, rules)
-
-    if check:
-        # make sure there are no differences between _sync and _sync_check
-        for source_dir in source_dirs:
+    for dir in source_dirs:
+        output_dir = f"{dir[0]}_sync_check/" if check else dir[1]
+        subprocess.check_call(["black", "--target-version=py38", output_dir])
+        subprocess.check_call(["isort", output_dir])
+        for file in glob("*.py", root_dir=dir[0]):
+            # remove asyncio from sync files
             subprocess.check_call(
-                ["black", "--target-version=py38", f"{source_dir}/_sync_check/"]
+                ["sed", "-i.bak", "/^import asyncio$/d", f"{output_dir}{file}"]
             )
-            subprocess.check_call(["isort", f"{source_dir}/_sync_check/"])
             subprocess.check_call(
                 [
-                    "diff",
-                    "-x",
-                    "__pycache__",
-                    f"{source_dir}/_sync",
-                    f"{source_dir}/_sync_check",
+                    "sed",
+                    "-i.bak",
+                    "s/asyncio\\.run(main())/main()/",
+                    f"{output_dir}{file}",
                 ]
             )
-            subprocess.check_call(["rm", "-rf", f"{source_dir}/_sync_check"])
+            subprocess.check_call(["rm", f"{output_dir}{file}.bak"])
+
+            if check:
+                # make sure there are no differences between _sync and _sync_check
+                subprocess.check_call(
+                    [
+                        "diff",
+                        f"{dir[1]}{file}",
+                        f"{output_dir}{file}",
+                    ]
+                )
+
+        if check:
+            subprocess.check_call(["rm", "-rf", output_dir])
 
 
 if __name__ == "__main__":