Added generator to observable

Author: syrusakbary

graphql/execution/executors/asyncio.py

@@ -25,7 +25,15 @@ def ensure_future(coro_or_future, loop=None):
                 del task._source_traceback[-1]
             return task
         else:
-            raise TypeError('A Future, a coroutine or an awaitable is required')
+            raise TypeError(
+                'A Future, a coroutine or an awaitable is required')
+
+try:
+    from .asyncio_utils import asyncgen_to_observable, isasyncgen
+except:
+    def isasyncgen(obj): False
+
+    def asyncgen_to_observable(asyncgen): pass
 
 
 class AsyncioExecutor(object):
@@ -50,4 +58,6 @@ def execute(self, fn, *args, **kwargs):
             future = ensure_future(result, loop=self.loop)
             self.futures.append(future)
             return Promise.resolve(future)
+        elif isasyncgen(result):
+            return asyncgen_to_observable(result)
         return result

graphql/execution/executors/asyncio_utils.py

@@ -0,0 +1,18 @@
+from inspect import isasyncgen
+from asyncio import ensure_future
+from rx import Observable
+
+
+def asyncgen_to_observable(asyncgen):
+    def emit(observer):
+        ensure_future(iterate_asyncgen(asyncgen, observer))
+    return Observable.create(emit)
+
+
+async def iterate_asyncgen(asyncgen, observer):
+    try:
+        async for item in asyncgen:
+            observer.on_next(item)
+        observer.on_completed()
+    except Exception as e:
+        observer.on_error(e)