Await resolvers in parallel (#11)

Author: mawa

graphql/execution/execute.py

@@ -1,3 +1,4 @@
+from asyncio import gather
 from inspect import isawaitable
 from typing import (
     Any,
@@ -437,12 +438,17 @@ def execute_fields(
         # Otherwise, results is a map from field name to the result of
         # resolving that field, which is possibly a coroutine object.
         # Return a coroutine object that will yield this same map, but with
-        # any coroutines awaited and replaced with the values they yielded.
+        # any coroutines awaited in parallel and replaced with the values they
+        # yielded.
         async def get_results():
-            return {
-                key: await value if isawaitable(value) else value
-                for key, value in results.items()
-            }
+            async def await_kv(key, value):
+                return key, await value if isawaitable(value) else value
+
+            pairs = await gather(
+                *(await_kv(key, value) for key, value in results.items())
+            )
+
+            return dict(pairs)
 
         return get_results()
 

tests/execution/test_executor.py

@@ -20,6 +20,7 @@
     GraphQLResolveInfo,
     ResponsePath,
 )
+from .util import compare_query_results_unordered
 
 
 def describe_execute_handles_basic_execution_tasks():
@@ -417,67 +418,70 @@ async def asyncReturnErrorWithExtensions(self, _info):
             )
         )
 
-        assert await execute(schema, ast, Data()) == (
-            {
-                "syncOk": "sync ok",
-                "syncError": None,
-                "syncRawError": None,
-                "syncReturnError": None,
-                "syncReturnErrorList": ["sync0", None, "sync2", None],
-                "asyncOk": "async ok",
-                "asyncError": None,
-                "asyncRawError": None,
-                "asyncReturnError": None,
-                "asyncReturnErrorWithExtensions": None,
-            },
-            [
-                {
-                    "message": "Error getting syncError",
-                    "locations": [(3, 15)],
-                    "path": ["syncError"],
-                },
-                {
-                    "message": "Error getting syncRawError",
-                    "locations": [(4, 15)],
-                    "path": ["syncRawError"],
-                },
-                {
-                    "message": "Error getting syncReturnError",
-                    "locations": [(5, 15)],
-                    "path": ["syncReturnError"],
-                },
-                {
-                    "message": "Error getting syncReturnErrorList1",
-                    "locations": [(6, 15)],
-                    "path": ["syncReturnErrorList", 1],
-                },
-                {
-                    "message": "Error getting syncReturnErrorList3",
-                    "locations": [(6, 15)],
-                    "path": ["syncReturnErrorList", 3],
-                },
-                {
-                    "message": "Error getting asyncError",
-                    "locations": [(8, 15)],
-                    "path": ["asyncError"],
-                },
-                {
-                    "message": "Error getting asyncRawError",
-                    "locations": [(9, 15)],
-                    "path": ["asyncRawError"],
-                },
-                {
-                    "message": "Error getting asyncReturnError",
-                    "locations": [(10, 15)],
-                    "path": ["asyncReturnError"],
-                },
+        compare_query_results_unordered(
+            await execute(schema, ast, Data()),
+            (
                 {
-                    "message": "Error getting asyncReturnErrorWithExtensions",
-                    "locations": [(11, 15)],
-                    "path": ["asyncReturnErrorWithExtensions"],
-                    "extensions": {"foo": "bar"},
+                    "syncOk": "sync ok",
+                    "syncError": None,
+                    "syncRawError": None,
+                    "syncReturnError": None,
+                    "syncReturnErrorList": ["sync0", None, "sync2", None],
+                    "asyncOk": "async ok",
+                    "asyncError": None,
+                    "asyncRawError": None,
+                    "asyncReturnError": None,
+                    "asyncReturnErrorWithExtensions": None,
                 },
-            ],
+                [
+                    {
+                        "message": "Error getting syncError",
+                        "locations": [(3, 15)],
+                        "path": ["syncError"],
+                    },
+                    {
+                        "message": "Error getting syncRawError",
+                        "locations": [(4, 15)],
+                        "path": ["syncRawError"],
+                    },
+                    {
+                        "message": "Error getting syncReturnError",
+                        "locations": [(5, 15)],
+                        "path": ["syncReturnError"],
+                    },
+                    {
+                        "message": "Error getting syncReturnErrorList1",
+                        "locations": [(6, 15)],
+                        "path": ["syncReturnErrorList", 1],
+                    },
+                    {
+                        "message": "Error getting syncReturnErrorList3",
+                        "locations": [(6, 15)],
+                        "path": ["syncReturnErrorList", 3],
+                    },
+                    {
+                        "message": "Error getting asyncError",
+                        "locations": [(8, 15)],
+                        "path": ["asyncError"],
+                    },
+                    {
+                        "message": "Error getting asyncRawError",
+                        "locations": [(9, 15)],
+                        "path": ["asyncRawError"],
+                    },
+                    {
+                        "message": "Error getting asyncReturnError",
+                        "locations": [(10, 15)],
+                        "path": ["asyncReturnError"],
+                    },
+                    {
+                        "message": "Error getting asyncReturnErrorWithExtensions",
+                        "locations": [(11, 15)],
+                        "path": ["asyncReturnErrorWithExtensions"],
+                        "extensions": {"foo": "bar"},
+                    },
+                ],
+            ),
         )
 
     def full_response_path_is_included_for_non_nullable_fields():
@@ -866,3 +870,42 @@ def resolve_field(self, parent_type, source, field_nodes, path):
             {"foo": "barbar"},
             None,
         )
+
+    @mark.asyncio
+    async def resolve_fields_in_parallel():
+        class Barrier(object):
+            # Makes progress only if at least `count` callers are `wait()`ing.
+            def __init__(self, count):
+                self.ev = asyncio.Event()
+                self.count = count
+
+            async def wait(self):
+                self.count -= 1
+                if self.count == 0:
+                    self.ev.set()
+
+                return await self.ev.wait()
+
+        barrier = Barrier(2)
+
+        async def f(*args):
+            return await barrier.wait()
+
+        schema = GraphQLSchema(
+            GraphQLObjectType(
+                "Object",
+                {
+                    "foo": GraphQLField(GraphQLBoolean, resolve=f),
+                    "bar": GraphQLField(GraphQLBoolean, resolve=f),
+                },
+            )
+        )
+
+        query = "{foo, bar}"
+        ast = parse(query)
+
+        res = await asyncio.wait_for(
+            execute(schema, ast), 1.0  # don't wait forever for the test to fail
+        )
+
+        assert res == ({"foo": True, "bar": True}, None)