GraphQL-core is a Python port of GraphQL.js, the JavaScript reference implementation for GraphQL.
—
High-level functions that provide the complete GraphQL execution pipeline including parsing, validation, and execution in a single call. These are the primary entry points for executing GraphQL operations.
Execute GraphQL operations asynchronously, supporting async resolvers and middleware with complete error handling and context management.
async def graphql(
schema: GraphQLSchema,
source: Union[str, Source],
root_value: Any = None,
context_value: Any = None,
variable_values: Optional[Dict[str, Any]] = None,
operation_name: Optional[str] = None,
field_resolver: Optional[GraphQLFieldResolver] = None,
type_resolver: Optional[GraphQLTypeResolver] = None,
middleware: Optional[Middleware] = None,
execution_context_class: Optional[Type[ExecutionContext]] = None,
is_awaitable: Optional[Callable[[Any], bool]] = None,
) -> ExecutionResultParameters:
schema: The GraphQL schema to use for validation and executionsource: GraphQL query string or Source objectroot_value: Initial value passed to root-level resolverscontext_value: Context object available to all resolvers via resolve_info.contextvariable_values: Variable values for parameterized queriesoperation_name: Operation name when document contains multiple operationsfield_resolver: Default field resolver functiontype_resolver: Default type resolver for abstract typesmiddleware: Middleware to wrap resolver executionexecution_context_class: Custom execution context classis_awaitable: Function to determine if values are awaitableReturns: ExecutionResult with data, errors, and extensions
import asyncio
from graphql import GraphQLSchema, GraphQLObjectType, GraphQLField, GraphQLString, graphql
schema = GraphQLSchema(
query=GraphQLObjectType(
name='Query',
fields={
'hello': GraphQLField(
GraphQLString,
resolve=lambda obj, info: 'Hello World!'
)
}
)
)
async def main():
result = await graphql(schema, '{ hello }')
print(result.data) # {'hello': 'Hello World!'}
asyncio.run(main())Execute GraphQL operations synchronously, ensuring all resolvers complete without async behavior for simpler integration patterns.
def graphql_sync(
schema: GraphQLSchema,
source: Union[str, Source],
root_value: Any = None,
context_value: Any = None,
variable_values: Optional[Dict[str, Any]] = None,
operation_name: Optional[str] = None,
field_resolver: Optional[GraphQLFieldResolver] = None,
type_resolver: Optional[GraphQLTypeResolver] = None,
middleware: Optional[Middleware] = None,
execution_context_class: Optional[Type[ExecutionContext]] = None,
check_sync: bool = False,
) -> ExecutionResultParameters: Same as graphql() except:
check_sync: If True, validates that no awaitable values are returned from resolversReturns: ExecutionResult with data, errors, and extensions
from graphql import GraphQLSchema, GraphQLObjectType, GraphQLField, GraphQLString, graphql_sync
schema = GraphQLSchema(
query=GraphQLObjectType(
name='Query',
fields={
'greeting': GraphQLField(
GraphQLString,
resolve=lambda obj, info, name='World': f'Hello {name}!'
)
}
)
)
# Simple query
result = graphql_sync(schema, '{ greeting }')
print(result.data) # {'greeting': 'Hello World!'}
# Query with variables
result = graphql_sync(
schema,
'query($name: String) { greeting(name: $name) }',
variable_values={'name': 'GraphQL'}
)
print(result.data) # {'greeting': 'Hello GraphQL!'}All execution functions return ExecutionResult objects containing the query results and any errors encountered.
class ExecutionResult:
data: Optional[Dict[str, Any]]
errors: Optional[List[GraphQLError]]
extensions: Optional[Dict[str, Any]]
def __init__(
self,
data: Optional[Dict[str, Any]] = None,
errors: Optional[List[GraphQLError]] = None,
extensions: Optional[Dict[str, Any]] = None,
)
class FormattedExecutionResult(TypedDict):
data: Optional[Dict[str, Any]]
errors: Optional[List[GraphQLFormattedError]]
extensions: Optional[Dict[str, Any]]Properties:
data: Query result data as nested dictionaries, None if execution failederrors: List of GraphQL errors encountered during executionextensions: Additional metadata or debugging informationfrom graphql import graphql_sync, GraphQLSchema, GraphQLObjectType, GraphQLField, GraphQLString
def error_resolver(obj, info):
raise Exception("Something went wrong!")
schema = GraphQLSchema(
query=GraphQLObjectType(
name='Query',
fields={
'error': GraphQLField(GraphQLString, resolve=error_resolver),
'success': GraphQLField(GraphQLString, resolve=lambda obj, info: 'OK')
}
)
)
result = graphql_sync(schema, '{ error success }')
print(result.data) # {'error': None, 'success': 'OK'}
print(len(result.errors)) # 1
print(result.errors[0].message) # "Something went wrong!"# Import required types
from typing import Any, Dict, List, Optional, Union, Callable, Type
from graphql.type import GraphQLSchema, GraphQLFieldResolver, GraphQLTypeResolver
from graphql.execution import ExecutionContext, Middleware
from graphql.language import Source
from graphql.error import GraphQLError
from graphql.pyutils import AwaitableOrValue
# Core execution result type
ExecutionResult = class ExecutionResult:
data: Optional[Dict[str, Any]]
errors: Optional[List[GraphQLError]]
extensions: Optional[Dict[str, Any]]
# Formatted result for serialization
FormattedExecutionResult = TypedDict('FormattedExecutionResult', {
'data': Optional[Dict[str, Any]],
'errors': Optional[List[GraphQLFormattedError]],
'extensions': Optional[Dict[str, Any]]
})
# Union type for potentially awaitable values
AwaitableOrValue = Union[T, Awaitable[T]]Install with Tessl CLI
npx tessl i tessl/pypi-graphql-core