Add GraphQLSchema.get_root_type and deprecate get_operation_root_type

Replicates graphql/graphql-js@96b146d

Author: Cito

src/graphql/__init__.py

@@ -37,7 +37,6 @@
     Path,
     Undefined,
 )
-from ..utilities.get_operation_root_type import get_operation_root_type
 from ..type import (
     GraphQLAbstractType,
     GraphQLField,
@@ -340,12 +339,19 @@ def execute_operation(
 
         Implements the "Executing operations" section of the spec.
         """
-        type_ = get_operation_root_type(self.schema, operation)
-        fields = collect_fields(
+        root_type = self.schema.get_root_type(operation.operation)
+        if root_type is None:
+            raise GraphQLError(
+                "Schema is not configured to execute"
+                f" {operation.operation.value} operation.",
+                operation,
+            )
+
+        root_fields = collect_fields(
             self.schema,
             self.fragments,
             self.variable_values,
-            type_,
+            root_type,
             operation.selection_set,
         )
 
@@ -360,7 +366,7 @@ def execute_operation(
                 self.execute_fields_serially
                 if operation.operation == OperationType.MUTATION
                 else self.execute_fields
-            )(type_, root_value, path, fields)
+            )(root_type, root_value, path, root_fields)
         except GraphQLError as error:
             self.errors.append(error)
             return None

src/graphql/execution/execute.py

@@ -21,7 +21,6 @@
 from ..language import DocumentNode
 from ..pyutils import Path, inspect
 from ..type import GraphQLFieldResolver, GraphQLSchema
-from ..utilities import get_operation_root_type
 from .map_async_iterator import MapAsyncIterator
 
 __all__ = ["subscribe", "create_source_event_stream"]
@@ -163,25 +162,32 @@ async def create_source_event_stream(
 
 async def execute_subscription(context: ExecutionContext) -> AsyncIterable[Any]:
     schema = context.schema
-    type_ = get_operation_root_type(schema, context.operation)
-    fields = collect_fields(
+
+    root_type = schema.subscription_type
+    if root_type is None:
+        raise GraphQLError(
+            "Schema is not configured to execute subscription operation.",
+            context.operation,
+        )
+
+    root_fields = collect_fields(
         schema,
         context.fragments,
         context.variable_values,
-        type_,
+        root_type,
         context.operation.selection_set,
     )
-    response_name, field_nodes = next(iter(fields.items()))
-    field_def = get_field_def(schema, type_, field_nodes[0])
+    response_name, field_nodes = next(iter(root_fields.items()))
+    field_def = get_field_def(schema, root_type, field_nodes[0])
 
     if not field_def:
         field_name = field_nodes[0].name.value
         raise GraphQLError(
             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)
+    path = Path(None, response_name, root_type.name)
+    info = context.build_resolve_info(field_def, field_nodes, root_type, path)
 
     # Implements the "ResolveFieldEventStream" algorithm from GraphQL specification.
     # It differs from "ResolveFieldValue" due to providing a different `resolveFn`.

src/graphql/execution/subscribe.py

@@ -12,7 +12,7 @@
 )
 
 from ..error import GraphQLError
-from ..language import ast
+from ..language import ast, OperationType
 from ..pyutils import inspect, is_collection, is_description, FrozenList
 from .definition import (
     GraphQLAbstractType,
@@ -326,6 +326,9 @@ def __deepcopy__(self, memo_: Dict) -> "GraphQLSchema":
             assume_valid=True,
         )
 
+    def get_root_type(self, operation: OperationType) -> Optional[GraphQLObjectType]:
+        return getattr(self, f"{operation.value}_type")
+
     def get_type(self, name: str) -> Optional[GraphQLNamedType]:
         return self.type_map.get(name)
 

src/graphql/type/schema.py

@@ -15,7 +15,11 @@ def get_operation_root_type(
     schema: GraphQLSchema,
     operation: Union[OperationDefinitionNode, OperationTypeDefinitionNode],
 ) -> GraphQLObjectType:
-    """Extract the root type of the operation from the schema."""
+    """Extract the root type of the operation from the schema.
+
+    .. deprecated:: 3.2
+       Please use `GraphQLSchema.getRootType` instead. Will be removed in v3.3.
+    """
     operation_type = operation.operation
     if operation_type == OperationType.QUERY:
         query_type = schema.query_type

src/graphql/utilities/get_operation_root_type.py

@@ -166,8 +166,8 @@ def enter_directive(self, node: DirectiveNode) -> None:
         self._directive = self._schema.get_directive(node.name.value)
 
     def enter_operation_definition(self, node: OperationDefinitionNode) -> None:
-        type_ = getattr(self._schema, f"{node.operation.value}_type")
-        self._type_stack.append(type_ if is_object_type(type_) else None)
+        root_type = self._schema.get_root_type(node.operation)
+        self._type_stack.append(root_type if is_object_type(root_type) else None)
 
     def enter_inline_fragment(self, node: InlineFragmentNode) -> None:
         type_condition_ast = node.type_condition

src/graphql/utilities/type_info.py

@@ -789,6 +789,35 @@ class Data:
         result = execute_sync(schema, document, Data(), operation_name="S")
         assert result == ({"a": "b"}, None)
 
+    def resolves_to_an_error_if_schema_does_not_support_operation():
+        schema = GraphQLSchema(assume_valid=True)
+
+        document = parse(
+            """
+            query Q { __typename }
+            mutation M { __typename }
+            subscription S { __typename }
+            """
+        )
+
+        with raises(
+            GraphQLError,
+            match=r"^Schema is not configured to execute query operation\.",
+        ):
+            execute_sync(schema, document, operation_name="Q")
+
+        with raises(
+            GraphQLError,
+            match=r"^Schema is not configured to execute mutation operation\.",
+        ):
+            execute_sync(schema, document, operation_name="M")
+
+        with raises(
+            GraphQLError,
+            match=r"^Schema is not configured to execute subscription operation\.",
+        ):
+            execute_sync(schema, document, operation_name="S")
+
     @mark.asyncio
     async def correct_field_ordering_despite_execution_order():
         schema = GraphQLSchema(

tests/execution/test_executor.py

@@ -296,6 +296,24 @@ async def throws_an_error_if_some_of_required_arguments_are_missing():
         with raises(TypeError, match="missing .* positional argument: 'document'"):
             await subscribe(schema=schema)  # type: ignore
 
+    @mark.asyncio
+    async def resolves_to_an_error_if_schema_does_not_support_subscriptions():
+        schema = GraphQLSchema(query=DummyQueryType)
+        document = parse("subscription { unknownField }")
+
+        result = await subscribe(schema, document)
+
+        assert result == (
+            None,
+            [
+                {
+                    "message": "Schema is not configured to execute"
+                    " subscription operation.",
+                    "locations": [(1, 1)],
+                }
+            ],
+        )
+
     @mark.asyncio
     async def resolves_to_an_error_for_unknown_subscription_field():
         schema = GraphQLSchema(

tests/execution/test_subscribe.py

@@ -27,7 +27,7 @@ def get_operation_node(doc: DocumentNode) -> OperationDefinitionNode:
     return operation_node
 
 
-def describe_get_operation_root_type():
+def describe_deprecated_get_operation_root_type():
     def gets_a_query_type_for_an_unnamed_operation_definition_node():
         test_schema = GraphQLSchema(query_type)
         doc = parse("{ field }")