execute/subscribe: simplify to improve debugging experience

Replicates graphql/graphql-js@cadcef8

Author: Cito

src/graphql/error/located_error.py

@@ -1,5 +1,6 @@
 from typing import TYPE_CHECKING, Collection, Optional, Union
 
+from ..pyutils import inspect
 from .graphql_error import GraphQLError
 
 if TYPE_CHECKING:
@@ -9,7 +10,7 @@
 
 
 def located_error(
-    original_error: Union[Exception, GraphQLError],
+    original_error: Exception,
     nodes: Optional[Union["None", Collection["Node"]]],
     path: Optional[Collection[Union[str, int]]] = None,
 ) -> GraphQLError:
@@ -19,8 +20,9 @@ def located_error(
     GraphQL operation, produce a new GraphQLError aware of the location in the document
     responsible for the original Exception.
     """
+    # Sometimes a non-error is thrown, wrap it as a TypeError to ensure consistency.
     if not isinstance(original_error, Exception):
-        raise TypeError("Expected an Exception.")
+        original_error = TypeError(f"Unexpected error value: {inspect(original_error)}")
     # Note: this uses a brand-check to support GraphQL errors originating from
     # other contexts.
     if isinstance(original_error, GraphQLError) and original_error.path is not None:

src/graphql/execution/execute.py

@@ -331,8 +331,6 @@ def execute_operation(
         # Errors from sub-fields of a NonNull type may propagate to the top level, at
         # which point we still log the error and null the parent field, which in this
         # case is the entire response.
-        #
-        # Similar to complete_value_catching_error.
         try:
             # noinspection PyArgumentList
             result = (
@@ -595,6 +593,7 @@ def resolve_field(
         if not field_def:
             return Undefined
 
+        return_type = field_def.type
         resolve_fn = field_def.resolve or self.field_resolver
 
         if self.middleware_manager:
@@ -604,29 +603,6 @@ def resolve_field(
 
         # Get the resolve function, regardless of if its result is normal or abrupt
         # (error).
-        result = self.resolve_field_value_or_error(
-            field_def, field_nodes, resolve_fn, source, info
-        )
-
-        return self.complete_value_catching_error(
-            field_def.type, field_nodes, info, path, result
-        )
-
-    def resolve_field_value_or_error(
-        self,
-        field_def: GraphQLField,
-        field_nodes: List[FieldNode],
-        resolve_fn: GraphQLFieldResolver,
-        source: Any,
-        info: GraphQLResolveInfo,
-    ) -> Union[Exception, Any]:
-        """Resolve field to a value or an error.
-
-        Isolates the "ReturnOrAbrupt" behavior to not de-opt the resolve_field()
-        method. Returns the result of resolveFn or the abrupt-return Error object.
-
-        For internal use only.
-        """
         try:
             # Build a dictionary of arguments from the field.arguments AST, using the
             # variables scope to fulfill any variable references.
@@ -635,58 +611,38 @@ def resolve_field_value_or_error(
             # Note that contrary to the JavaScript implementation, we pass the context
             # value as part of the resolve info.
             result = resolve_fn(source, info, **args)
+
+            completed: AwaitableOrValue[Any]
             if self.is_awaitable(result):
                 # noinspection PyShadowingNames
                 async def await_result() -> Any:
                     try:
-                        return await result
+                        completed = self.complete_value(
+                            return_type, field_nodes, info, path, await result
+                        )
+                        if self.is_awaitable(completed):
+                            return await completed
+                        return completed
                     except Exception as error:
-                        return error
+                        self.handle_field_error(error, field_nodes, path, return_type)
+                        return None
 
                 return await_result()
-            return result
-        except Exception as error:
-            return error
 
-    def complete_value_catching_error(
-        self,
-        return_type: GraphQLOutputType,
-        field_nodes: List[FieldNode],
-        info: GraphQLResolveInfo,
-        path: Path,
-        result: Any,
-    ) -> AwaitableOrValue[Any]:
-        """Complete a value while catching an error.
-
-        This is a small wrapper around completeValue which detects and logs errors in
-        the execution context.
-        """
-        completed: AwaitableOrValue[Any]
-        try:
-            if self.is_awaitable(result):
-
-                async def await_result() -> Any:
-                    value = self.complete_value(
-                        return_type, field_nodes, info, path, await result
-                    )
-                    if self.is_awaitable(value):
-                        return await value
-                    return value
-
-                completed = await_result()
-            else:
-                completed = self.complete_value(
-                    return_type, field_nodes, info, path, result
-                )
+            completed = self.complete_value(
+                return_type, field_nodes, info, path, result
+            )
             if self.is_awaitable(completed):
                 # noinspection PyShadowingNames
                 async def await_completed() -> Any:
                     try:
                         return await completed
                     except Exception as error:
                         self.handle_field_error(error, field_nodes, path, return_type)
+                        return None
 
                 return await_completed()
+
             return completed
         except Exception as error:
             self.handle_field_error(error, field_nodes, path, return_type)
@@ -825,10 +781,45 @@ def complete_list_value(
         for index, item in enumerate(result):
             # No need to modify the info object containing the path, since from here on
             # it is not ever accessed by resolver functions.
-            field_path = path.add_key(index, None)
-            completed_item = self.complete_value_catching_error(
-                item_type, field_nodes, info, field_path, item
-            )
+            item_path = path.add_key(index, None)
+            completed_item: AwaitableOrValue[Any]
+            if is_awaitable(item):
+                # noinspection PyShadowingNames
+                async def await_completed(item: Any, item_path: Path) -> Any:
+                    try:
+                        completed = self.complete_value(
+                            item_type, field_nodes, info, item_path, await item
+                        )
+                        if is_awaitable(completed):
+                            return await completed
+                        return completed
+                    except Exception as error:
+                        self.handle_field_error(
+                            error, field_nodes, item_path, item_type
+                        )
+                        return None
+
+                completed_item = await_completed(item, item_path)
+            else:
+                try:
+                    completed_item = self.complete_value(
+                        item_type, field_nodes, info, item_path, item
+                    )
+                    if is_awaitable(completed_item):
+                        # noinspection PyShadowingNames
+                        async def await_completed(item: Any, item_path: Path) -> Any:
+                            try:
+                                return await item
+                            except Exception as error:
+                                self.handle_field_error(
+                                    error, field_nodes, item_path, item_type
+                                )
+                                return None
+
+                        completed_item = await_completed(completed_item, item_path)
+                except Exception as error:
+                    self.handle_field_error(error, field_nodes, item_path, item_type)
+                    completed_item = None
 
             if is_awaitable(completed_item):
                 append_awaitable(index)

src/graphql/subscription/subscribe.py

@@ -3,11 +3,9 @@
     Any,
     AsyncIterable,
     AsyncIterator,
-    Awaitable,
     Dict,
     Optional,
     Union,
-    cast,
 )
 
 from ..error import GraphQLError, located_error
@@ -18,6 +16,7 @@
     ExecutionContext,
     ExecutionResult,
 )
+from ..execution.values import get_argument_values
 from ..language import DocumentNode
 from ..pyutils import Path, inspect
 from ..type import GraphQLFieldResolver, GraphQLSchema
@@ -56,18 +55,15 @@ async def subscribe(
     If the operation succeeded, the coroutine will yield an AsyncIterator, which yields
     a stream of ExecutionResults representing the response stream.
     """
-    try:
-        result_or_stream = await create_source_event_stream(
-            schema,
-            document,
-            root_value,
-            context_value,
-            variable_values,
-            operation_name,
-            subscribe_field_resolver,
-        )
-    except GraphQLError as error:
-        return ExecutionResult(data=None, errors=[error])
+    result_or_stream = await create_source_event_stream(
+        schema,
+        document,
+        root_value,
+        context_value,
+        variable_values,
+        operation_name,
+        subscribe_field_resolver,
+    )
     if isinstance(result_or_stream, ExecutionResult):
         return result_or_stream
 
@@ -111,11 +107,15 @@ async def create_source_event_stream(
 
     Returns a coroutine that yields an AsyncIterable.
 
-    If the client provided invalid arguments, the source stream could not be created,
-    or the resolver did not return an AsyncIterable, this function will throw an error,
-    which should be caught and handled by the caller.
+    If the client-provided arguments to this function do not result in a compliant
+    subscription, a GraphQL Response (ExecutionResult) with descriptive errors and no
+    data will be returned.
 
-    A Source Event Stream represents a sequence of events, each of which triggers a
+    If the source stream could not be created due to faulty subscription resolver logic
+    or underlying systems, the coroutine object will yield a single ExecutionResult
+    containing ``errors`` and no ``data``.
+
+    A source event stream represents a sequence of events, each of which triggers a
     GraphQL execution for that event.
 
     This may be useful when hosting the stateful subscription service in a different
@@ -143,6 +143,14 @@ async def create_source_event_stream(
     if isinstance(context, list):
         return ExecutionResult(data=None, errors=context)
 
+    try:
+        return await execute_subscription(context)
+    except GraphQLError as error:
+        return ExecutionResult(data=None, errors=[error])
+
+
+async def execute_subscription(context: ExecutionContext) -> AsyncIterable[Any]:
+    schema = context.schema
     type_ = get_operation_root_type(schema, context.operation)
     fields = context.collect_fields(type_, context.operation.selection_set, {}, set())
     response_names = list(fields)
@@ -157,29 +165,35 @@ async def create_source_event_stream(
             f"The subscription field '{field_name}' is not defined.", field_nodes
         )
 
+    path = Path(None, response_name, type_.name)
+    info = context.build_resolve_info(field_def, field_nodes, type_, path)
+
+    # Implements the "ResolveFieldEventStream" algorithm from GraphQL specification.
+    # It differs from "ResolveFieldValue" due to providing a different `resolveFn`.
+
+    # Build a dictionary of arguments from the field.arguments AST, using the
+    # variables scope to fulfill any variable references.
+    args = get_argument_values(field_def, field_nodes[0], context.variable_values)
+
     # Call the `subscribe()` resolver or the default resolver to produce an
     # AsyncIterable yielding raw payloads.
     resolve_fn = field_def.subscribe or context.field_resolver
 
-    path = Path(None, response_name, type_.name)
-
-    info = context.build_resolve_info(field_def, field_nodes, type_, path)
+    try:
+        event_stream = resolve_fn(context.root_value, info, **args)
+        if context.is_awaitable(event_stream):
+            event_stream = await event_stream
+    except Exception as error:
+        event_stream = error
 
-    # `resolve_field_value_or_error` implements the "ResolveFieldEventStream" algorithm
-    # from GraphQL specification. It differs from `resolve_field_value` due to
-    # providing a different `resolve_fn`.
-    result = context.resolve_field_value_or_error(
-        field_def, field_nodes, resolve_fn, root_value, info
-    )
-    event_stream = await cast(Awaitable, result) if isawaitable(result) else result
-    # If `event_stream` is an Error, rethrow a located error.
     if isinstance(event_stream, Exception):
         raise located_error(event_stream, field_nodes, path.as_list())
 
     # Assert field returned an event stream, otherwise yield an error.
-    if isinstance(event_stream, AsyncIterable):
-        return event_stream
-    raise TypeError(
-        "Subscription field must return AsyncIterable."
-        f" Received: {inspect(event_stream)}."
-    )
+    if not isinstance(event_stream, AsyncIterable):
+        raise TypeError(
+            "Subscription field must return AsyncIterable."
+            f" Received: {inspect(event_stream)}."
+        )
+
+    return event_stream

tests/error/test_located_error.py

@@ -1,16 +1,13 @@
 from typing import cast, Any
 
-from pytest import raises  # type: ignore
-
 from graphql.error import GraphQLError, located_error
 
 
 def describe_located_error():
     def throws_without_an_original_error():
-        with raises(TypeError) as exc_info:
-            # noinspection PyTypeChecker
-            located_error([], [], [])  # type: ignore
-        assert str(exc_info.value) == "Expected an Exception."
+        e = located_error([], [], []).original_error  # type: ignore
+        assert isinstance(e, TypeError)
+        assert str(e) == "Unexpected error value: []"
 
     def passes_graphql_error_through():
         path = ["path", 3, "to", "field"]