Use type guards (#183)

Author: Cito

pyproject.toml

@@ -42,7 +42,7 @@ Changelog = "https://github.com/graphql-python/graphql-core/releases"
 [tool.poetry.dependencies]
 python = "^3.7"
 typing-extensions = [
-    { version = "^4.3", python = "<3.8" }
+    { version = "^4.4", python = "<3.10" }
 ]
 
 [tool.poetry.group.test]
@@ -138,7 +138,7 @@ module = [
 disallow_untyped_defs = false
 
 [tool.pytest.ini_options]
-minversion = "7.1"
+minversion = "7.2"
 # Only run benchmarks as tests.
 # To actually run the benchmarks, use --benchmark-enable on the command line.
 # To run the slow tests (fuzzing), add --run-slow on the command line.

src/graphql/execution/collect_fields.py

@@ -1,5 +1,5 @@
 from collections import defaultdict
-from typing import Any, Dict, List, Set, Union, cast
+from typing import Any, Dict, List, Set, Union
 
 from ..language import (
     FieldNode,
@@ -9,7 +9,6 @@
     SelectionSetNode,
 )
 from ..type import (
-    GraphQLAbstractType,
     GraphQLIncludeDirective,
     GraphQLObjectType,
     GraphQLSchema,
@@ -166,7 +165,7 @@ def does_fragment_condition_match(
     if conditional_type is type_:
         return True
     if is_abstract_type(conditional_type):
-        return schema.is_sub_type(cast(GraphQLAbstractType, conditional_type), type_)
+        return schema.is_sub_type(conditional_type, type_)
     return False
 
 

src/graphql/execution/execute.py

@@ -21,6 +21,10 @@
     from typing import TypedDict
 except ImportError:  # Python < 3.8
     from typing_extensions import TypedDict
+try:
+    from typing import TypeGuard
+except ImportError:  # Python < 3.10
+    from typing_extensions import TypeGuard
 
 from ..error import GraphQLError, GraphQLFormattedError, located_error
 from ..language import (
@@ -39,7 +43,6 @@
     GraphQLFieldResolver,
     GraphQLLeafType,
     GraphQLList,
-    GraphQLNonNull,
     GraphQLObjectType,
     GraphQLOutputType,
     GraphQLResolveInfo,
@@ -187,7 +190,9 @@ class ExecutionContext:
     errors: List[GraphQLError]
     middleware_manager: Optional[MiddlewareManager]
 
-    is_awaitable = staticmethod(default_is_awaitable)
+    is_awaitable: Callable[[Any], TypeGuard[Awaitable]] = staticmethod(
+        default_is_awaitable  # type: ignore
+    )
 
     def __init__(
         self,
@@ -607,7 +612,7 @@ def complete_value(
         # result is null.
         if is_non_null_type(return_type):
             completed = self.complete_value(
-                cast(GraphQLNonNull, return_type).of_type,
+                return_type.of_type,
                 field_nodes,
                 info,
                 path,
@@ -627,25 +632,25 @@ def complete_value(
         # If field type is List, complete each item in the list with inner type
         if is_list_type(return_type):
             return self.complete_list_value(
-                cast(GraphQLList, return_type), field_nodes, info, path, result
+                return_type, field_nodes, info, path, result
             )
 
         # If field type is a leaf type, Scalar or Enum, serialize to a valid value,
         # returning null if serialization is not possible.
         if is_leaf_type(return_type):
-            return self.complete_leaf_value(cast(GraphQLLeafType, return_type), result)
+            return self.complete_leaf_value(return_type, result)
 
         # If field type is an abstract type, Interface or Union, determine the runtime
         # Object type and complete for that type.
         if is_abstract_type(return_type):
             return self.complete_abstract_value(
-                cast(GraphQLAbstractType, return_type), field_nodes, info, path, result
+                return_type, field_nodes, info, path, result
             )
 
         # If field type is Object, execute and complete all sub-selections.
         if is_object_type(return_type):
             return self.complete_object_value(
-                cast(GraphQLObjectType, return_type), field_nodes, info, path, result
+                return_type, field_nodes, info, path, result
             )
 
         # Not reachable. All possible output types have been considered.
@@ -684,7 +689,6 @@ async def async_iterable_to_list(
                 "Expected Iterable, but did not find one for field"
                 f" '{info.parent_type.name}.{info.field_name}'."
             )
-        result = cast(Iterable[Any], result)
 
         # This is specified as a simple map, however we're optimizing the path where
         # the list contains no coroutine objects by avoiding creating another coroutine
@@ -876,8 +880,6 @@ def ensure_valid_runtime_type(
                 field_nodes,
             )
 
-        runtime_type = cast(GraphQLObjectType, runtime_type)
-
         if not self.schema.is_sub_type(return_type, runtime_type):
             raise GraphQLError(
                 f"Runtime Object type '{runtime_type.name}' is not a possible"

src/graphql/execution/values.py

@@ -1,4 +1,4 @@
-from typing import Any, Callable, Collection, Dict, List, Optional, Union, cast
+from typing import Any, Callable, Collection, Dict, List, Optional, Union
 
 from ..error import GraphQLError
 from ..language import (
@@ -21,7 +21,6 @@
 from ..type import (
     GraphQLDirective,
     GraphQLField,
-    GraphQLInputType,
     GraphQLSchema,
     is_input_type,
     is_non_null_type,
@@ -92,7 +91,6 @@ def coerce_variable_values(
             )
             continue
 
-        var_type = cast(GraphQLInputType, var_type)
         if var_name not in inputs:
             if var_def_node.default_value:
                 coerced_values[var_name] = value_from_ast(

src/graphql/language/parser.py

@@ -201,9 +201,8 @@ def __init__(
         no_location: bool = False,
         allow_legacy_fragment_variables: bool = False,
     ):
-        source = (
-            cast(Source, source) if is_source(source) else Source(cast(str, source))
-        )
+        if not is_source(source):
+            source = Source(cast(str, source))
 
         self._lexer = Lexer(source)
         self._no_location = no_location

src/graphql/language/predicates.py

@@ -1,3 +1,5 @@
+from typing import Union
+
 from .ast import (
     DefinitionNode,
     ExecutableDefinitionNode,
@@ -15,6 +17,12 @@
 )
 
 
+try:
+    from typing import TypeGuard
+except ImportError:  # Python < 3.10
+    from typing_extensions import TypeGuard
+
+
 __all__ = [
     "is_definition_node",
     "is_executable_definition_node",
@@ -29,27 +37,27 @@
 ]
 
 
-def is_definition_node(node: Node) -> bool:
+def is_definition_node(node: Node) -> TypeGuard[DefinitionNode]:
     """Check whether the given node represents a definition."""
     return isinstance(node, DefinitionNode)
 
 
-def is_executable_definition_node(node: Node) -> bool:
+def is_executable_definition_node(node: Node) -> TypeGuard[ExecutableDefinitionNode]:
     """Check whether the given node represents an executable definition."""
     return isinstance(node, ExecutableDefinitionNode)
 
 
-def is_selection_node(node: Node) -> bool:
+def is_selection_node(node: Node) -> TypeGuard[SelectionNode]:
     """Check whether the given node represents a selection."""
     return isinstance(node, SelectionNode)
 
 
-def is_value_node(node: Node) -> bool:
+def is_value_node(node: Node) -> TypeGuard[ValueNode]:
     """Check whether the given node represents a value."""
     return isinstance(node, ValueNode)
 
 
-def is_const_value_node(node: Node) -> bool:
+def is_const_value_node(node: Node) -> TypeGuard[ValueNode]:
     """Check whether the given node represents a constant value."""
     return is_value_node(node) and (
         any(is_const_value_node(value) for value in node.values)
@@ -60,26 +68,28 @@ def is_const_value_node(node: Node) -> bool:
     )
 
 
-def is_type_node(node: Node) -> bool:
+def is_type_node(node: Node) -> TypeGuard[TypeNode]:
     """Check whether the given node represents a type."""
     return isinstance(node, TypeNode)
 
 
-def is_type_system_definition_node(node: Node) -> bool:
+def is_type_system_definition_node(node: Node) -> TypeGuard[TypeSystemDefinitionNode]:
     """Check whether the given node represents a type system definition."""
     return isinstance(node, TypeSystemDefinitionNode)
 
 
-def is_type_definition_node(node: Node) -> bool:
+def is_type_definition_node(node: Node) -> TypeGuard[TypeDefinitionNode]:
     """Check whether the given node represents a type definition."""
     return isinstance(node, TypeDefinitionNode)
 
 
-def is_type_system_extension_node(node: Node) -> bool:
+def is_type_system_extension_node(
+    node: Node,
+) -> TypeGuard[Union[SchemaExtensionNode, TypeExtensionNode]]:
     """Check whether the given node represents a type system extension."""
     return isinstance(node, (SchemaExtensionNode, TypeExtensionNode))
 
 
-def is_type_extension_node(node: Node) -> bool:
+def is_type_extension_node(node: Node) -> TypeGuard[TypeExtensionNode]:
     """Check whether the given node represents a type extension."""
     return isinstance(node, TypeExtensionNode)

src/graphql/language/source.py

@@ -3,6 +3,12 @@
 from .location import SourceLocation
 
 
+try:
+    from typing import TypeGuard
+except ImportError:  # Python < 3.10
+    from typing_extensions import TypeGuard
+
+
 __all__ = ["Source", "is_source"]
 
 DEFAULT_NAME = "GraphQL request"
@@ -66,7 +72,7 @@ def __ne__(self, other: Any) -> bool:
         return not self == other
 
 
-def is_source(source: Any) -> bool:
+def is_source(source: Any) -> TypeGuard[Source]:
     """Test if the given value is a Source object.
 
     For internal use only.

src/graphql/pyutils/is_awaitable.py

@@ -1,14 +1,20 @@
 import inspect
 from types import CoroutineType, GeneratorType
-from typing import Any
+from typing import Any, Awaitable
+
+
+try:
+    from typing import TypeGuard
+except ImportError:  # Python < 3.10
+    from typing_extensions import TypeGuard
 
 
 __all__ = ["is_awaitable"]
 
 CO_ITERABLE_COROUTINE = inspect.CO_ITERABLE_COROUTINE
 
 
-def is_awaitable(value: Any) -> bool:
+def is_awaitable(value: Any) -> TypeGuard[Awaitable]:
     """Return true if object can be passed to an ``await`` expression.
 
     Instead of testing if the object is an instance of abc.Awaitable, it checks
@@ -18,7 +24,7 @@ def is_awaitable(value: Any) -> bool:
         # check for coroutine objects
         isinstance(value, CoroutineType)
         # check for old-style generator based coroutine objects
-        or isinstance(value, GeneratorType)
+        or isinstance(value, GeneratorType)  # for Python < 3.11
         and bool(value.gi_code.co_flags & CO_ITERABLE_COROUTINE)
         # check for other awaitables (e.g. futures)
         or hasattr(value, "__await__")

src/graphql/pyutils/is_iterable.py

@@ -2,6 +2,12 @@
 from typing import Any, ByteString, Collection, Iterable, Mapping, Text, ValuesView
 
 
+try:
+    from typing import TypeGuard
+except ImportError:  # Python < 3.10
+    from typing_extensions import TypeGuard
+
+
 __all__ = ["is_collection", "is_iterable"]
 
 collection_types: Any = [Collection]
@@ -16,14 +22,14 @@
 not_iterable_types: Any = (ByteString, Mapping, Text)
 
 
-def is_collection(value: Any) -> bool:
+def is_collection(value: Any) -> TypeGuard[Collection]:
     """Check if value is a collection, but not a string or a mapping."""
     return isinstance(value, collection_types) and not isinstance(
         value, not_iterable_types
     )
 
 
-def is_iterable(value: Any) -> bool:
+def is_iterable(value: Any) -> TypeGuard[Iterable]:
     """Check if value is an iterable, but not a string or a mapping."""
     return isinstance(value, iterable_types) and not isinstance(
         value, not_iterable_types