Execute with promises

Author: syrusakbary

graphql/execution/__init__.py

@@ -20,42 +20,6 @@
 """
 from .execute import execute as _execute
 from .base import ExecutionResult
-# from .executor import Executor
-# from .middlewares.sync import SynchronousExecutionMiddleware
 
 def execute(schema, root, ast, operation_name='', args=None):
     return _execute(schema, ast, root, variable_values=args, operation_name=operation_name)
-
-# def execute(schema, root, ast, operation_name='', args=None):
-#     """
-#     Executes an AST synchronously. Assumes that the AST is already validated.
-#     """
-#     return get_default_executor().execute(schema, ast, root, args, operation_name, validate_ast=False)
-
-
-# _default_executor = None
-
-
-# def get_default_executor():
-#     """
-#         Gets the default executor to be used in the `execute` function above.
-#     """
-#     global _default_executor
-#     if _default_executor is None:
-#         _default_executor = Executor([SynchronousExecutionMiddleware()])
-
-#     return _default_executor
-
-
-# def set_default_executor(executor):
-#     """
-#         Sets the default executor to be used in the `execute` function above.
-
-#         If passed `None` will reset to the original default synchronous executor.
-#     """
-#     assert isinstance(executor, Executor) or executor is None
-#     global _default_executor
-#     _default_executor = executor
-
-
-# __all__ = ['ExecutionResult', 'Executor', 'execute', 'get_default_executor', 'set_default_executor']

graphql/execution/base.py

@@ -19,9 +19,9 @@ class ExecutionContext(object):
     and the fragments defined in the query document"""
 
     __slots__ = 'schema', 'fragments', 'root_value', 'operation', 'variable_values', 'errors', 'context_value', \
-                'argument_values_cache'
+                'argument_values_cache', 'executor'
 
-    def __init__(self, schema, document_ast, root_value, context_value, variable_values, operation_name):
+    def __init__(self, schema, document_ast, root_value, context_value, variable_values, operation_name, executor):
         """Constructs a ExecutionContext object from the arguments passed
         to execute, which we will pass throughout the other execution
         methods."""
@@ -63,6 +63,7 @@ def __init__(self, schema, document_ast, root_value, context_value, variable_val
         self.errors = errors
         self.context_value = context_value
         self.argument_values_cache = {}
+        self.executor = executor
 
     def get_argument_values(self, field_def, field_ast):
         k = field_def, field_ast

graphql/execution/execute.py

@@ -16,20 +16,24 @@
 from .base import (ExecutionContext, ExecutionResult, ResolveInfo, Undefined,
                    collect_fields, default_resolve_fn, get_field_def,
                    get_operation_root_type)
+from .executors.sync import SyncExecutor
 
 
-
-def execute(schema, document_ast, root_value=None, context_value=None, variable_values=None, operation_name=None):
+def execute(schema, document_ast, root_value=None, context_value=None, variable_values=None, operation_name=None, executor=None):
     assert schema, 'Must provide schema'
     assert isinstance(schema, GraphQLSchema), 'Schema must be an instance of GraphQLSchema. Also ensure that there are not multiple versions of GraphQL installed in your node_modules directory.'
 
+    if executor is None:
+        executor = SyncExecutor()
+
     context = ExecutionContext(
         schema,
         document_ast,
         root_value,
         context_value,
         variable_values,
-        operation_name
+        operation_name,
+        executor
     )
 
     def executor(resolve, reject):
@@ -42,7 +46,10 @@ def on_rejected(error):
     def on_resolve(data):
         return ExecutionResult(data=data, errors=context.errors)
 
-    return Promise(executor).catch(on_rejected).then(on_resolve).value
+    p = Promise(executor).catch(on_rejected).then(on_resolve)
+    context.executor.wait_until_finished()
+    return p.value
+
 
 
 def execute_operation(exe_context, operation, root_value):
@@ -177,7 +184,7 @@ def resolve_field(exe_context, parent_type, source, field_asts):
 def resolve_or_error(resolve_fn, source, args, exe_context, info):
     try:
         # return resolve_fn(source, args, exe_context, info)
-        return resolve_fn(source, args, info)
+        return exe_context.executor.execute(resolve_fn, source, args, info)
     except Exception as e:
         return e
 

graphql/execution/executors/__init__.py

@@ -0,0 +1,34 @@
+from __future__ import absolute_import
+
+from asyncio import Future, ensure_future, iscoroutine, get_event_loop, wait
+from graphql.pyutils.aplus import Promise
+
+
+def process_future_result(promise):
+    def handle_future_result(future):
+        exception = future.exception()
+        if exception:
+            promise.reject(exception)
+        else:
+            promise.fulfill(future.result())
+
+    return handle_future_result
+
+
+class AsyncioExecutor(object):
+    def __init__(self):
+        self.loop = get_event_loop()
+        self.futures = []
+
+    def wait_until_finished(self):
+        self.loop.run_until_complete(wait(self.futures))
+
+    def execute(self, fn, *args, **kwargs):
+        result = fn(*args, **kwargs)
+        if isinstance(result, Future) or iscoroutine(result):
+            promise = Promise()
+            future = ensure_future(result)
+            self.futures.append(future)
+            future.add_done_callback(process_future_result(promise))
+            return promise
+        return result

graphql/execution/executors/asyncio.py

@@ -0,0 +1,20 @@
+from __future__ import absolute_import
+
+import gevent
+from .utils import process
+from ...pyutils.aplus import Promise
+
+
+class GeventExecutor(object):
+    def __init__(self):
+        self.jobs = []
+
+    def wait_until_finished(self):
+        [j.join() for j in self.jobs]
+        # gevent.joinall(self.jobs)
+
+    def execute(self, fn, *args, **kwargs):
+        promise = Promise()
+        job = gevent.spawn(process, promise, fn, args, kwargs)
+        self.jobs.append(job)
+        return promise

graphql/execution/executors/gevent.py

@@ -0,0 +1,28 @@
+from multiprocessing import Pool, Process, Queue
+from ...pyutils.aplus import Promise
+from .utils import process
+
+def queue_process(q):
+    promise, fn, args, kwargs = q.get()
+    process(promise, fn, args, kwargs)
+
+
+class ProcessExecutor(object):
+    def __init__(self):
+        self.processes = []
+        self.q = Queue()
+
+    def wait_until_finished(self):
+        for _process in self.processes:
+            _process.join()
+        self.q.close()
+        self.q.join_thread()
+
+    def execute(self, fn, *args, **kwargs):
+        promise = Promise()
+        
+        q.put([promise, fn, args, kwargs], False)
+        _process = Process(target=queue_process, args=(self.q))
+        _process.start()
+        self.processes.append(_process)
+        return promise

graphql/execution/executors/process.py

@@ -0,0 +1,6 @@
+class SyncExecutor(object):
+    def wait_until_finished(self):
+        pass
+
+    def execute(self, fn, *args, **kwargs):
+        return fn(*args, **kwargs)

graphql/execution/executors/sync.py

@@ -0,0 +1,19 @@
+from threading import Thread
+from ...pyutils.aplus import Promise
+from .utils import process
+
+
+class ThreadExecutor(object):
+    def __init__(self):
+        self.threads = []
+
+    def wait_until_finished(self):
+        for thread in self.threads:
+            thread.join()
+
+    def execute(self, fn, *args, **kwargs):
+        promise = Promise()
+        thread = Thread(target=process, args=(promise, fn, args, kwargs))
+        thread.start()
+        self.threads.append(thread)
+        return promise

graphql/execution/executors/thread.py

@@ -0,0 +1,6 @@
+def process(p, f, args, kwargs):
+    try:
+        val = f(*args, **kwargs)
+        p.fulfill(val)
+    except Exception as e:
+        p.reject(e)