checkpoint

Author: sethmlarson

dev-requirements.txt

@@ -1,6 +1,7 @@
 requests>=2, <3
 pytest
 pytest-cov
+pytest-asyncio
 coverage
 mock
 nosexcover

elasticsearch/_async/client/__init__.py

@@ -308,7 +308,7 @@ async def clear_scroll(self, body=None, scroll_id=None, params=None, headers=Non
         elif scroll_id:
             params["scroll_id"] = scroll_id
 
-        return self.transport.perform_request(
+        return await self.transport.perform_request(
             "DELETE", "/_search/scroll", params=params, headers=headers, body=body
         )
 
@@ -421,7 +421,7 @@ async def create(self, index, id, body, doc_type=None, params=None, headers=None
         else:
             path = _make_path(index, doc_type, id)
 
-        return self.transport.perform_request(
+        return await self.transport.perform_request(
             "POST" if id in SKIP_IN_PATH else "PUT",
             path,
             params=params,
@@ -1001,7 +1001,7 @@ async def index(self, index, body, id=None, params=None, headers=None):
             if param in SKIP_IN_PATH:
                 raise ValueError("Empty value passed for a required argument.")
 
-        return self.transport.perform_request(
+        return await self.transport.perform_request(
             "POST" if id in SKIP_IN_PATH else "PUT",
             _make_path(index, "_doc", id),
             params=params,
@@ -1417,7 +1417,7 @@ async def scroll(self, body=None, scroll_id=None, params=None, headers=None):
         elif scroll_id:
             params["scroll_id"] = scroll_id
 
-        return self.transport.perform_request(
+        return await self.transport.perform_request(
             "POST", "/_search/scroll", params=params, headers=headers, body=body
         )
 
@@ -1791,7 +1791,7 @@ async def update(self, index, id, body, doc_type=None, params=None, headers=None
         else:
             path = _make_path(index, doc_type, id, "_update")
 
-        return self.transport.perform_request(
+        return await self.transport.perform_request(
             "POST", path, params=params, headers=headers, body=body
         )
 

elasticsearch/_async/client/cluster.py

@@ -133,7 +133,7 @@ async def stats(self, node_id=None, params=None, headers=None):
             false)
         :arg timeout: Explicit operation timeout
         """
-        return self.transport.perform_request(
+        return await self.transport.perform_request(
             "GET",
             "/_cluster/stats"
             if node_id in SKIP_IN_PATH

elasticsearch/_async/compat.py

@@ -19,3 +19,46 @@ def get_running_loop():
         if not loop.is_running():
             raise RuntimeError("no running event loop")
         return loop
+
+
+def get_sleep():
+    loop = get_running_loop()
+
+    async def sleep(duration):
+        await asyncio.sleep(duration, loop=loop)
+    return sleep
+
+
+def azip(*iterables):
+    print("AZIP", iterables)
+    iterators = [aiter(x) for x in iterables]
+    print("AZIPTOR", iterators)
+
+    async def generator():
+        while True:
+            try:
+                tuple_items = []
+                for iterator in iterators:
+                    tuple_items.append(await iterator.__anext__())
+                print("azip tuple", tuple_items)
+                yield tuple(tuple_items)
+            except StopAsyncIteration:
+                break
+
+    return generator().__aiter__()
+
+
+def aiter(x):
+    """Creates an async iterator out of async or sync iterables
+    and iterators. Map the 'aiter' token to 'iter'
+    """
+    if hasattr(x, "__aiter__"):
+        return x.__aiter__()
+    elif hasattr(x, "__anext__"):
+        return x
+
+    async def aiter_wrapper():
+        for item in x:
+            yield item
+
+    return aiter_wrapper().__aiter__()