tessl/pypi-graphql-core

GraphQL-core is a Python port of GraphQL.js, the JavaScript reference implementation for GraphQL.

—

Pending

Overview

Eval results

Files

Query Execution Engine

Name: tessl/pypi-graphql-core
Author: tessl

Execute validated GraphQL operations with resolver functions, middleware support, subscription handling, and comprehensive value processing. Provides the core execution engine for GraphQL operations after parsing and validation.

Capabilities

Core Execution Functions

Execute GraphQL operations with full control over execution parameters and behavior.

def execute(
    schema: GraphQLSchema,
    document: DocumentNode,
    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,
    subscribe_field_resolver: Optional[GraphQLFieldResolver] = None,
    middleware: Optional[Middleware] = None,
    execution_context_class: Optional[Type[ExecutionContext]] = None,
    is_awaitable: Optional[Callable[[Any], bool]] = None,
) -> AwaitableOrValue[ExecutionResult]

def execute_sync(
    schema: GraphQLSchema,
    document: DocumentNode,
    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,
) -> ExecutionResult

Parameters:

schema: GraphQL schema to execute against
document: Parsed and validated GraphQL document
root_value: Initial value passed to root resolvers
context_value: Context object passed to all resolvers
variable_values: Variable values for the operation
operation_name: Specific operation to execute (if document has multiple)
field_resolver: Default field resolver function
type_resolver: Default type resolver for abstract types
subscribe_field_resolver: Special resolver for subscription fields (execute only)
check_sync: Whether to check that no awaitable values are returned (execute_sync only)
middleware: Middleware stack for resolver wrapping
execution_context_class: Custom execution context class
is_awaitable: Function to determine if values are awaitable

Returns: ExecutionResult or awaitable ExecutionResult

Usage Example

from graphql import execute, parse, build_schema
import asyncio

schema = build_schema('''
    type Query {
        user(id: ID!): User
        posts: [Post]
    }
    
    type User {
        id: ID!
        name: String
    }
    
    type Post {
        id: ID!
        title: String
        author: User
    }
''')

async def resolve_user(obj, info, id):
    return {'id': id, 'name': f'User {id}'}

async def resolve_posts(obj, info):
    return [
        {'id': '1', 'title': 'First Post'},
        {'id': '2', 'title': 'Second Post'}
    ]

async def resolve_post_author(obj, info):
    return {'id': '1', 'name': 'Author Name'}

# Set resolvers on schema
schema.query_type.fields['user'].resolve = resolve_user
schema.query_type.fields['posts'].resolve = resolve_posts
schema.type_map['Post'].fields['author'].resolve = resolve_post_author

async def main():
    document = parse('{ user(id: "123") { name } posts { title author { name } } }')
    result = await execute(schema, document)
    print(result.data)

asyncio.run(main())

Default Resolvers

Built-in resolver functions for common resolution patterns.

def default_field_resolver(source: Any, info: GraphQLResolveInfo, **args) -> Any
def default_type_resolver(value: Any, info: GraphQLResolveInfo, abstract_type: GraphQLAbstractType) -> Union[str, GraphQLObjectType, None]

default_field_resolver behavior:

If field name exists as attribute, return getattr(source, field_name)
If field name exists as dictionary key, return source[field_name]
If source has a method matching field name, call it with args
Otherwise return None

Custom Resolver Example

from graphql import default_field_resolver

def custom_field_resolver(source, info, **args):
    # Custom logic before default resolution
    if info.field_name == 'secret' and not info.context.get('authenticated'):
        return None
    
    # Fall back to default behavior
    result = default_field_resolver(source, info, **args)
    
    # Custom logic after default resolution
    if isinstance(result, str):
        return result.upper()
    
    return result

# Use custom resolver
result = execute(schema, document, field_resolver=custom_field_resolver)

Execution Context

Manage execution state and provide access to execution information.

class ExecutionContext:
    def __init__(
        self,
        schema: GraphQLSchema,
        fragments: Dict[str, FragmentDefinitionNode],
        root_value: Any,
        context_value: Any,
        operation: OperationDefinitionNode,
        variable_values: Dict[str, Any],
        field_resolver: GraphQLFieldResolver,
        type_resolver: GraphQLTypeResolver,
        subscribe_field_resolver: Optional[GraphQLFieldResolver],
        errors: List[GraphQLError],
        middleware_manager: Optional[MiddlewareManager],
        is_awaitable: Callable[[Any], bool],
    )
    
    schema: GraphQLSchema
    fragments: Dict[str, FragmentDefinitionNode]
    root_value: Any
    context_value: Any
    operation: OperationDefinitionNode
    variable_values: Dict[str, Any]
    field_resolver: GraphQLFieldResolver
    type_resolver: GraphQLTypeResolver
    errors: List[GraphQLError]
    
    def add_error(self, error: GraphQLError) -> None

Execution Results

Structured results containing data, errors, and extensions.

class ExecutionResult:
    def __init__(
        self,
        data: Optional[Dict[str, Any]] = None,
        errors: Optional[List[GraphQLError]] = None,
        extensions: Optional[Dict[str, Any]] = None,
    )
    
    data: Optional[Dict[str, Any]]
    errors: Optional[List[GraphQLError]]
    extensions: Optional[Dict[str, Any]]

class FormattedExecutionResult(TypedDict):
    data: Optional[Dict[str, Any]]
    errors: Optional[List[GraphQLFormattedError]]
    extensions: Optional[Dict[str, Any]]

Middleware Support

Wrap resolver execution with middleware for authentication, logging, caching, and other cross-cutting concerns.

class MiddlewareManager:
    def __init__(self, *middlewares: Middleware)
    
    def get_field_resolver(self, field_resolver: GraphQLFieldResolver) -> GraphQLFieldResolver

# Middleware type
Middleware = Union[
    Callable[[GraphQLFieldResolver], GraphQLFieldResolver],
    Callable[[Any, GraphQLResolveInfo, GraphQLFieldResolver, **Any], Any],
    object  # Object with resolve method
]

Middleware Example

from graphql import MiddlewareManager

class AuthMiddleware:
    def resolve(self, next_resolver, root, info, **args):
        if info.field_name == 'secret' and not info.context.get('user'):
            raise Exception('Authentication required')
        return next_resolver(root, info, **args)

class LoggingMiddleware:
    def resolve(self, next_resolver, root, info, **args):
        print(f"Resolving {info.parent_type.name}.{info.field_name}")
        result = next_resolver(root, info, **args)
        print(f"Resolved {info.parent_type.name}.{info.field_name}: {result}")
        return result

# Apply middleware
middleware = MiddlewareManager(AuthMiddleware(), LoggingMiddleware())
result = execute(schema, document, middleware=middleware)

Subscription Support

Execute GraphQL subscriptions with event streams and async iteration.

def subscribe(
    schema: GraphQLSchema,
    document: DocumentNode,
    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,
    subscribe_field_resolver: Optional[GraphQLFieldResolver] = None,
) -> AwaitableOrValue[Union[ExecutionResult, AsyncIterator[ExecutionResult]]]

def create_source_event_stream(
    schema: GraphQLSchema,
    document: DocumentNode,
    root_value: Any = None,
    context_value: Any = None,
    variable_values: Optional[Dict[str, Any]] = None,
    operation_name: Optional[str] = None,
    subscribe_field_resolver: Optional[GraphQLFieldResolver] = None,
) -> AwaitableOrValue[Union[AsyncIterator[Any], ExecutionResult]]

class MapAsyncIterator:
    def __init__(self, async_iterable: AsyncIterable[Any], map_fn: Callable[[Any], Any])
    
    def __aiter__(self) -> AsyncIterator[Any]
    async def __anext__(self) -> Any

Subscription Example

from graphql import subscribe, parse, build_schema
import asyncio
from asyncio import Queue

schema = build_schema('''
    type Subscription {
        messageAdded: String
    }
    
    type Query {
        dummy: String
    }
''')

message_queue = Queue()

async def resolve_message_added(obj, info):
    # Return async iterator
    async def message_generator():
        while True:
            message = await message_queue.get()
            yield message
    
    return message_generator()

schema.subscription_type.fields['messageAdded'].subscribe = resolve_message_added

async def main():
    subscription = parse('subscription { messageAdded }')
    
    # Start subscription
    subscription_result = await subscribe(schema, subscription)
    
    if hasattr(subscription_result, '__aiter__'):
        # Handle async iterator
        async for result in subscription_result:
            print(f"Received: {result.data}")
    else:
        # Handle error result
        print(f"Subscription error: {subscription_result.errors}")

# Send events
async def send_message(message):
    await message_queue.put(message)

asyncio.run(main())

Value Processing

Process arguments, variables, and directive values during execution.

def get_argument_values(
    arg_defs: Dict[str, GraphQLArgument],
    arg_nodes: Optional[List[ArgumentNode]],
    variable_values: Optional[Dict[str, Any]] = None,
) -> Dict[str, Any]

def get_directive_values(
    directive_def: GraphQLDirective,
    directive_node: Optional[DirectiveNode],
    variable_values: Optional[Dict[str, Any]] = None,
) -> Optional[Dict[str, Any]]

def get_variable_values(
    schema: GraphQLSchema,
    var_def_nodes: List[VariableDefinitionNode],
    inputs: Optional[Dict[str, Any]],
    max_errors: Optional[int] = None,
) -> CoercedVariableValues

class CoercedVariableValues:
    values: Dict[str, Any]
    errors: List[GraphQLError]

Value Processing Example

from graphql import get_argument_values, GraphQLArgument, GraphQLString

# Define arguments
arg_defs = {
    'name': GraphQLArgument(GraphQLString, default_value='World'),
    'greeting': GraphQLArgument(GraphQLString, default_value='Hello')
}

# Parse argument nodes (would come from AST)
# arg_nodes = [ArgumentNode(name=NameNode(value='name'), value=StringValueNode(value='GraphQL'))]

# Get processed argument values
args = get_argument_values(arg_defs, arg_nodes, variable_values={})
print(args)  # {'name': 'GraphQL', 'greeting': 'Hello'}

Types

# Import required types
from typing import Any, Dict, List, Optional, Union, Callable, Type, AsyncIterator, AsyncIterable, Awaitable
from graphql.type import GraphQLSchema, GraphQLFieldResolver, GraphQLTypeResolver, GraphQLAbstractType, GraphQLArgument, GraphQLDirective
from graphql.language import DocumentNode, ArgumentNode, DirectiveNode, VariableDefinitionNode, FragmentDefinitionNode, OperationDefinitionNode
from graphql.error import GraphQLError, GraphQLFormattedError
from graphql.pyutils import AwaitableOrValue

# Core execution types
ExecutionResult = class ExecutionResult
FormattedExecutionResult = TypedDict
ExecutionContext = class ExecutionContext
MiddlewareManager = class MiddlewareManager

# Middleware and resolver types
Middleware = Union[
    Callable[[GraphQLFieldResolver], GraphQLFieldResolver],
    Callable[[Any, GraphQLResolveInfo, GraphQLFieldResolver], Any],
    object
]

GraphQLFieldResolver = Callable[..., Any]
GraphQLTypeResolver = Callable[..., Any]

# Subscription types
MapAsyncIterator = class MapAsyncIterator

# Value processing types
class CoercedVariableValues:
    values: Dict[str, Any]
    errors: List[GraphQLError]

# Resolve info type
class GraphQLResolveInfo:
    field_name: str
    field_nodes: List[FieldNode]
    return_type: GraphQLOutputType
    parent_type: GraphQLObjectType
    path: ResponsePath
    schema: GraphQLSchema
    fragments: Dict[str, FragmentDefinitionNode]
    root_value: Any
    operation: OperationDefinitionNode
    variable_values: Dict[str, Any]
    context: Any
    is_awaitable: Callable[[Any], bool]

Install with Tessl CLI