let graphql inherit all parse/validate/execute options

Replicates graphql/graphql-js@928a321

Author: Cito

src/graphql/graphql.py

@@ -17,15 +17,16 @@
 )
 
 if TYPE_CHECKING:
-    from collections.abc import AsyncIterable, Awaitable, Callable
+    from collections.abc import AsyncIterable, Awaitable, Callable, Collection
     from typing import TypeGuard
 
     from .pyutils import AbortSignal, AwaitableOrValue
+    from .validation import ASTValidationRule
 
 __all__ = ["graphql", "graphql_sync"]
 
 
-async def graphql(
+async def graphql(  # noqa: PLR0913
     schema: GraphQLSchema,
     source: str | Source,
     root_value: Any = None,
@@ -40,6 +41,11 @@ async def graphql(
     is_async_iterable: Callable[[Any], TypeGuard[AsyncIterable]] | None = None,
     hide_suggestions: bool = False,
     abort_signal: AbortSignal | None = None,
+    no_location: bool = False,
+    max_tokens: int | None = None,
+    experimental_fragment_arguments: bool = False,
+    rules: Collection[type[ASTValidationRule]] | None = None,
+    max_errors: int | None = None,
 ) -> ExecutionResult:
     """Execute a GraphQL operation asynchronously.
 
@@ -92,6 +98,17 @@ async def graphql(
     :arg abort_signal:
       A signal object that can be used to cancel execution, e.g. the signal of an
       :class:`~graphql.AbortController`
+    :arg no_location:
+      Parse option: do not include location information in the parsed document.
+    :arg max_tokens:
+      Parse option: the maximum number of tokens the document may contain.
+    :arg experimental_fragment_arguments:
+      Parse option: enable experimental support for fragment arguments.
+    :arg rules:
+      The validation rules to use when validating the query.
+      If not provided, the specified rules are used.
+    :arg max_errors:
+      The maximum number of validation errors to report before stopping.
     """
     # Always return asynchronously for a consistent API.
     result = graphql_impl(
@@ -109,6 +126,11 @@ async def graphql(
         is_async_iterable,
         hide_suggestions,
         abort_signal,
+        no_location,
+        max_tokens,
+        experimental_fragment_arguments,
+        rules,
+        max_errors,
     )
 
     if default_is_awaitable(result):
@@ -127,7 +149,7 @@ def assume_not_async_iterable(_value: Any) -> TypeGuard[AsyncIterable]:
     return False
 
 
-def graphql_sync(
+def graphql_sync(  # noqa: PLR0913
     schema: GraphQLSchema,
     source: str | Source,
     root_value: Any = None,
@@ -141,6 +163,11 @@ def graphql_sync(
     check_sync: bool = False,
     hide_suggestions: bool = False,
     abort_signal: AbortSignal | None = None,
+    no_location: bool = False,
+    max_tokens: int | None = None,
+    experimental_fragment_arguments: bool = False,
+    rules: Collection[type[ASTValidationRule]] | None = None,
+    max_errors: int | None = None,
 ) -> ExecutionResult:
     """Execute a GraphQL operation synchronously.
 
@@ -172,6 +199,11 @@ def graphql_sync(
         is_async_iterable,
         hide_suggestions,
         abort_signal,
+        no_location,
+        max_tokens,
+        experimental_fragment_arguments,
+        rules,
+        max_errors,
     )
 
     # Assert that the execution was synchronous.
@@ -183,7 +215,7 @@ def graphql_sync(
     return cast("ExecutionResult", result)
 
 
-def graphql_impl(
+def graphql_impl(  # noqa: PLR0913
     schema: GraphQLSchema,
     source: str | Source,
     root_value: Any,
@@ -198,6 +230,11 @@ def graphql_impl(
     is_async_iterable: Callable[[Any], TypeGuard[AsyncIterable]] | None = None,
     hide_suggestions: bool = False,
     abort_signal: AbortSignal | None = None,
+    no_location: bool = False,
+    max_tokens: int | None = None,
+    experimental_fragment_arguments: bool = False,
+    rules: Collection[type[ASTValidationRule]] | None = None,
+    max_errors: int | None = None,
 ) -> AwaitableOrValue[ExecutionResult]:
     """Execute a query, return asynchronously only if necessary."""
     # Validate Schema
@@ -206,15 +243,20 @@ def graphql_impl(
 
     # Parse
     try:
-        document = parse(source)
+        document = parse(
+            source,
+            no_location=no_location,
+            max_tokens=max_tokens,
+            experimental_fragment_arguments=experimental_fragment_arguments,
+        )
     except GraphQLError as error:
         return ExecutionResult(data=None, errors=[error])
 
     # Validate
     from .validation import validate
 
     if validation_errors := validate(
-        schema, document, hide_suggestions=hide_suggestions
+        schema, document, rules, max_errors, hide_suggestions=hide_suggestions
     ):
         return ExecutionResult(data=None, errors=validation_errors)
 

tests/test_graphql.py

@@ -0,0 +1,138 @@
+from asyncio import sleep
+
+import pytest
+
+from graphql import graphql, graphql_sync
+from graphql.error import GraphQLError
+from graphql.language import Source
+from graphql.type import (
+    GraphQLField,
+    GraphQLObjectType,
+    GraphQLSchema,
+    GraphQLString,
+)
+from graphql.validation import ValidationRule
+
+from .fixtures import cleanup
+
+pytestmark = [
+    pytest.mark.anyio,
+    pytest.mark.filterwarnings("ignore:coroutine .* was never awaited:RuntimeWarning"),
+]
+
+
+def _resolve_context_echo(_source, info):
+    return str(info.context)
+
+
+def _resolve_root_value(root_value, _info):
+    return root_value
+
+
+async def _resolve_root_value_async(root_value, _info):
+    return root_value
+
+
+schema = GraphQLSchema(
+    query=GraphQLObjectType(
+        "Query",
+        {
+            "a": GraphQLField(GraphQLString, resolve=lambda *_args: "A"),
+            "b": GraphQLField(GraphQLString, resolve=lambda *_args: "B"),
+            "contextEcho": GraphQLField(GraphQLString, resolve=_resolve_context_echo),
+            "syncField": GraphQLField(GraphQLString, resolve=_resolve_root_value),
+            "asyncField": GraphQLField(
+                GraphQLString, resolve=_resolve_root_value_async
+            ),
+        },
+    )
+)
+
+
+def describe_graphql():
+    async def passes_source_through_to_parse():
+        source = Source("{", "custom-query.graphql")
+
+        result = await graphql(schema, source)
+
+        assert result.errors
+        assert result.errors[0].source is not None
+        assert result.errors[0].source.name == "custom-query.graphql"
+
+    async def passes_rules_through_to_validate():
+        class CustomRule(ValidationRule):
+            def enter_field(self, node, *_args):
+                self.context.report_error(GraphQLError("custom rule error", node))
+
+        result = await graphql(schema, "{ a }", rules=[CustomRule])
+
+        assert result.errors
+        assert result.errors[0].message == "custom rule error"
+
+    async def passes_parse_options_through_to_parse():
+        class CustomRule(ValidationRule):
+            def enter_operation_definition(self, node, *_args):
+                message = "no location" if node.loc is None else "has location"
+                self.context.report_error(GraphQLError(message, node))
+
+        result = await graphql(
+            schema, "{ a }", no_location=True, rules=[CustomRule]
+        )
+
+        assert result.errors
+        assert result.errors[0].message == "no location"
+
+    async def passes_validation_options_through_to_validate():
+        result = await graphql(schema, "{ contextEho }", hide_suggestions=True)
+
+        assert result.errors
+        assert (
+            result.errors[0].message
+            == "Cannot query field 'contextEho' on type 'Query'."
+        )
+
+    async def passes_execution_args_through_to_execute():
+        result = await graphql(
+            schema,
+            """
+            query First {
+              a
+            }
+
+            query Second {
+              b
+            }
+            """,
+            operation_name="Second",
+        )
+
+        assert result == ({"b": "B"}, None)
+
+    async def returns_schema_validation_errors():
+        bad_schema = GraphQLSchema()
+        result = await graphql(bad_schema, "{ __typename }")
+
+        assert result.errors
+        assert result.errors[0].message == "Query root type must be provided."
+
+
+def describe_graphql_sync():
+    def returns_result_for_synchronous_execution():
+        result = graphql_sync(schema, "{ syncField }", root_value="rootValue")
+
+        assert result == ({"syncField": "rootValue"}, None)
+
+    async def throws_for_asynchronous_execution():
+        # Unlike graphql-js, graphql_sync only detects asynchronous execution when
+        # check_sync is set; without it, the coroutine is treated as a plain value.
+        with pytest.raises(RuntimeError) as exc_info:
+            graphql_sync(
+                schema, "{ asyncField }", root_value="rootValue", check_sync=True
+            )
+        assert (
+            str(exc_info.value)
+            == "GraphQL execution failed to complete synchronously."
+        )
+        del exc_info
+        await sleep(0)  # let the loop process the cancelled execution task
+        cleanup()