Types reference
Ariadne uses type annotations in its codebase.
Many parts of its API share or rely on common types, importable from ariadne.types module:
BaseProxyRootValue
class BaseProxyRootValue:
...
A RootValue wrapper that includes result JSON update logic.
Can be returned by the RootValue callable. Not used by Ariadne directly
but part of the support for Ariadne GraphQL Proxy.
Attributes
root_value: Optional[dict]:RootValueto use during query execution.
Constructor
def __init__(self, root_value: Optional[dict] = None):
...
Methods
update_result
def update_result(self, result: GraphQLResult) -> GraphQLResult:
...
An update function used to create a final GraphQL result tuple to
create a JSON response from.
Default implementation in BaseProxyRootValue is a passthrough that
returns result value without any changes.
ContextValue
ContextValue = Union[Any, Callable[[Any], Any], Callable[[Any, dict], Any]]
Type for context_value option of GraphQL servers.
Context value is accessible in GraphQL resolvers as context attribute of second argument:
def resolve_example(_, info: GraphQLResolveInfo):
info.context # Do something with context
Default context
Default context value passed to resolvers by Ariadne is a dictionary with single key, request, which contains the HTTP framework specific representation of HTTP request:
def resolve_example(_, info: GraphQLResolveInfo):
request = info.context["request"] # Get request from context
Dynamic context value
If context value is a callable, it will be evaluated at the beginning of query execution.
It's called with two arguments:
request: an representation of HTTP request specific to the framework used.
data: an unvalidated JSON which may be a valid GraphQL request payload.
Callable can return any value which will then be passed to resolvers. Some implementations (like ariadne.asgi.GraphQL) support this callable being asynchronous.
Here's an example callable that retrieves application settings from database, then uses them together with request to retrieve authenticated user. Finally it exposes all three of those to GraphQL resolvers:
from ariadne.asgi import GraphQL
from starlette.requests import Request
from .auth import get_authenticated_user
from .settings import get_db_settings
from .schema import schema
async def get_context_value(request: Request, _):
settings = await get_db_settings()
user = await get_authenticated_user(request, settings)
return {
"request"; request,
"settings": settings,
"user": user,
}
graphql_app = GraphQL(
schema,
context_value=get_context_value,
)
ErrorFormatter
ErrorFormatter = Callable[[GraphQLError, bool], dict]
Type for custom error formatters.
Error formatter is a function called by Ariadne to convert GraphQLError instance into JSON-serializable dict to return to client.
It receives two arguments:
GraphQLError: an error to serialize.
bool: a debug flag, which tells the formatter if it should (True) or shouldn't (False) include debugging information.
Extension
class Extension:
...
Base class for extensions.
Subclasses of this class should override default methods to run custom logic during Query execution.
Methods
request_started
def request_started(self, context: ContextValue) -> None:
...
Extension hook executed at request's start.
request_finished
def request_finished(self, context: ContextValue) -> None:
...
Extension hook executed at request's end.
resolve
def resolve(
self,
next_: Resolver,
obj: Any,
info: GraphQLResolveInfo,
**kwargs,
) -> Any:
...
Extension hook wrapping field's value resolution.
Arguments
next_: a resolver or next extension's resolve method.
obj: a Python data structure to resolve value from.
info: a GraphQLResolveInfo instance for executed resolver.
**kwargs: extra arguments from GraphQL to pass to resolver.
Example
resolve should handle both sync and async next_:
from inspect import iscoroutinefunction
from time import time
from ariadne.types import Extension, Resolver
from graphql import GraphQLResolveInfo
from graphql.pyutils import is_awaitable
class MyExtension(Extension):
def __init__(self):
self.paths = []
def resolve(
self, next_: Resolver, obj: Any, info: GraphQLResolveInfo, **kwargs
) -> Any:
path = ".".join(map(str, info.path.as_list()))
# Fast implementation for synchronous resolvers
if not iscoroutinefunction(next_):
start_time = time()
result = next_(obj, info, **kwargs)
self.paths.append((path, time() - start_time))
return result
# Create async closure for async `next_` that GraphQL
# query executor will handle for us.
async def async_my_extension():
start_time = time()
result = await next_(obj, info, **kwargs)
if is_awaitable(result):
result = await result
self.paths.append((path, time() - start_time))
return result
# GraphQL query executor will execute this closure for us
return async_my_extension()
has_errors
def has_errors(
self,
errors: List[GraphQLError],
context: ContextValue,
) -> None:
...
Extension hook executed when GraphQL encountered errors.
format
def format(self, context: ContextValue) -> Optional[dict]:
...
Extension hook executed to retrieve extra data to include in result's
extensions data.
ExtensionList
ExtensionList = Optional[List[Union[Type['Extension'], Callable[[], 'Extension']]]]
List of extensions to use during GraphQL query execution.
Extensions
Extensions = Union[Callable[[Any, Optional[ContextValue]], ExtensionList], ExtensionList]
Type of extensions option of GraphQL servers.
It's either a list of extensions (see ExtensionList), or callable that returns this list.
Callable is evaluated with two arguments:
Any: the HTTP framework specific representation of HTTP request.
Optional[ContextValue]: a context value for this request, or None.
GraphQLResult
GraphQLResult = Tuple[bool, dict]
Result type for graphql and graphql_sync functions.
It's a tuple of two elements:
bool: True when query was executed successfully (without any errors), False otherwise.
dict: JSON-serializable query result.
Middleware
Middleware = Callable[..., Any]
GraphQL middleware.
Due to limitations in Mypy this type is unspecific. More accurate type is:
Middleware = Callable[[Resolver, Any, GraphQLResolveInfo, KwArg[Any]], Any]
Each middleware is called with three positional arguments and any number of keyword arguments:
Resolver: executed field's resolver or next GraphQL middleware.
Any: a Python representation of GraphQL type to resolve value from.
GraphQLResolveInfo: an instance of GraphQL object holding data about currently executed GraphQL field, and context attribute.
KwArg[Any]: if currently executed GraphQL field has any arguments, their values will be passed as keyword arguments.
MiddlewareList
MiddlewareList = Optional[Sequence[Middleware]]
List of middlewares to use during GraphQL query execution.
Middlewares
Middlewares = Union[Callable[[Any, Optional[ContextValue]], MiddlewareList], MiddlewareList]
Type of middleware option of GraphQL servers.
It's either a list of middleware (see MiddlewareList), or callable that returns this list.
Callable is evaluated with two arguments:
Any: the HTTP framework specific representation of HTTP request.
Optional[ContextValue]: a context value for this request, or None.
OnComplete
OnComplete = Callable[[WebSocket, Operation], Any]
Type of on_complete option of GraphQL websocket servers.
Callback function evaluated when individual GraphQL subscription is completed.
Called with two arguments:
WebSocket: the HTTP framework specific representation of websocket connection.
Operation: an object with closed subscription's data.
OnConnect
OnConnect = Callable[[WebSocket, Any], Any]
Type of on_connect option of GraphQL websocket servers.
Callback function evaluated when new websocket connection was established. Usually used to update connection's scope with contents of initial message.
Called with two arguments:
WebSocket: the HTTP framework specific representation of websocket connection.
Any: a data sent in WebSocket message.
OnDisconnect
OnDisconnect = Callable[[WebSocket], Any]
Type of on_disconnect option of GraphQL websocket servers.
Callback function evaluated when websocket connection is closed.
Called with one arguments:
WebSocket: the HTTP framework specific representation of websocket connection.
OnOperation
OnOperation = Callable[[WebSocket, Operation], Any]
Type of on_operation option of GraphQL websocket servers.
Callback function evaluated when individual GraphQL subscription is initiated.
Called with two arguments:
WebSocket: the HTTP framework specific representation of websocket connection.
Operation: an object with initiated subscription's data.
Operation
class Operation:
...
Dataclass representing single active GraphQL operation.
QueryParser
QueryParser = Callable[[ContextValue, Dict[str, Any]], DocumentNode]
Type of query_parser option of GraphQL servers.
Enables customization of server's GraphQL parsing logic. If not set or None, default parser is used instead.
Default query parser
Default query parser used by Ariadne is parse function from the graphql package.
Custom parser
Custom parser is a function or callable accepting two arguments:
context_value: a context value specific to this GraphQL server.
data: a dict with validated GraphQL request data (contains query string, optionally also has operationName string or variables dictionary).
Parser is required to return DocumentNode with parsed query or raise GraphQLError when query is data invalid .
Asynchronous parsers are not supported.
Example parser
Below code defines custom parser that discards query string altogether, using predefined queries instead:
from graphql import GraphQLError, parse
class AllowedQueriesParser:
def __init__(self):
self._queries = {}
def __call__(self, _, data):
operation_name = data.get("operationName")
if not operation_name:
raise GraphQLError(
"Explicit 'operationName' is required by the server."
)
if operation_name not in self._queries:
raise GraphQLError(
f"Operation 'operation_name' is not supported by the server."
)
return self._queries[operation_name]
def add_query(self, operation_name: str, query: str):
self._queries[operation_name] = parse(query)
allowed_queries_parser = AllowedQueriesParser()
allowed_queries_parser.add_query(
"GetUsers",
"""
query GetUsers {
users {
id
name
}
}
"""
)
QueryValidator
class QueryValidator(Protocol):
...
Type of query_validator option of GraphQL servers.
Enables customization of server's GraphQL query validation logic. If not set or None,
default graphql.validate is used instead.
Custom validator
Custom validator is a function or callable accepting up to 5 arguments:
schema: GraphQLSchema for the graphql schema
document_ast: DocumentNode result of query parser
rules: optional list of AST validation rules of type ASTValidationRule
max_errors: optional maximum number of errors to return
type_info: optional type info, pending deprecation in graphql 3.3
Validator is required to return List[GraphQLError] which should be empty
if there were no errors found
Methods
__call__
def __call__(
self,
schema: GraphQLSchema,
document_ast: DocumentNode,
rules: Optional[Collection[Type[ASTValidationRule]]] = None,
max_errors: Optional[int] = None,
type_info: Optional[TypeInfo] = None,
) -> List[GraphQLError]:
...
Example validator
Below code defines custom validator that mutates same Document with to "memoize" validation (this works best in a tandem with a memoized parser for end-to-end memoization of parsing and validation):
from graphql import GraphQLError, validate
from graphql import GraphQLSchema, DocumentNode, GraphQLError
from graphql.utilities.type_info import TypeInfo
from graphql.validation.rules import ASTValidationRule
def memoized_queries_validator(
schema: GraphQLSchema,
document_ast: DocumentNode,
rules: Optional[Collection[Type[ASTValidationRule]]] = None,
max_errors: Optional[int] = None,
type_info: Optional[TypeInfo] = None,
) -> List[GraphQLError]:
past_validation = getattr(document_ast, "_past_validation", None)
if past_validation:
return past_validation
validation = validate(schema, document_ast, *args, **kwargs)
setattr(document_ast, "_past_validation", validation)
return validation
Resolver
Resolver = Callable[..., Any]
Type for resolver functions.
Due to limitations in Mypy this type is unspecific. More accurate type is:
Resolver = Callable[[Any, GraphQLResolveInfo, KwArg[Any]], Any]
Each resolver is called with two positional arguments and any number of keyword arguments:
Any: a Python representation of GraphQL type to resolve value from.
GraphQLResolveInfo: an instance of GraphQL object holding data about currently executed GraphQL field, and context attribute.
KwArg[Any]: if currently executed GraphQL field has any arguments, their values will be passed as keyword arguments.
RootValue
RootValue = Union[Any, Callable[[Optional[Any], DocumentNode], Any], Callable[[Optional[Any], Optional[str], Optional[dict], DocumentNode], Any]]
Type for root_value option of GraphQL servers.
"Root value" is a value passed to root resolvers (resolvers set on Query, Mutation and Subscription fields) as first argument.
Default root value
Ariadne doesn't define root value by default. First argument of root resolvers is None.
Dynamic root value
If root value is a callable, it will be evaluated at the beginning of query handling. It's called with four arguments:
context_value: a context value specific to this GraphQL server.
operation_name: a str with name of operation to execute (or None).
variables: a dict with variables to pass to query's resolvers (or None).
document: a DocumentNode with parsed GraphQL query.
Callable can return any value which then will be passed to root resolvers. Some implementations (like ariadne.asgi.GraphQL) support this callable being asynchronous.
SchemaBindable
class SchemaBindable(Protocol):
...
Base class for bindable types.
Subclasses should extend the bind_to_schema method with custom logic for
populating an instance of GraphQL schema with Python logic and values.
Methods
bind_to_schema
def bind_to_schema(self, schema: GraphQLSchema) -> None:
...
Binds this Schema[Bindable](/docs/bindables) instance to the instance of GraphQL schema.
Example
Example InputType bindable that sets Python names for fields of GraphQL input:
from ariadne import SchemaBindable
from graphql import GraphQLInputType
class InputType(SchemaBindable):
_name: str
_fields: dict[str, str]
def __init__(self, name: str, fields: dict[str, str] | None):
self._name = name
self._fields = fields or {}
def set_field_out_name(self, field: str, out_name: str):
self._fields[field] = out_name
def bind_to_schema(self, schema: GraphQLSchema) -> None:
graphql_type = schema.get_type(self._name)
if not graphql_type:
raise ValueError(
"Type %s is not defined in the schema" % self.name
)
if not isinstance(graphql_type, GraphQLInputType):
raise ValueError(
"%s is defined in the schema, but it is instance of %s (expected %s)"
% (self.name, type(graphql_type).__name__, GraphQLInputType.__name__)
)
for field, out_name in self._fields.items():
schema_field = graphql_type.fields.get(field)
if not schema_field:
raise ValueError(
"Type %s does not define the %s field" % (self.name, field)
)
schema_field.out_name = out_name
Usage:
from ariadne import QueryType, make_executable_schema
from .input_type import InputType
from .users.models import User
input_type = InputType(
"UserInput",
{
"fullName": "full_name",
"yearOfBirth": "year_of_birth",
}
)
query_type = QueryType()
@query_type.field("countUsers")
def resolve_count_users(*_, input):
qs = User.objects
if input:
if input["full_name"]:
qs = qs.filter(full_name__ilike=input["full_name"])
if input["year_of_birth"]:
qs = qs.filter(dob__year=input["year_of_birth"])
return qs.count()
schema = make_executable_schema(
"""
type Query {
countUsers(input: UserInput!): Int!
}
input UserInput {
fullName: String
yearOfBirth: Int
}
""",
query_type,
input_type,
)
Subscriber
Subscriber = Callable[..., AsyncGenerator]
Type for subscription source functions.
Due to limitations in Mypy this type is unspecific. More accurate type is:
Subscriber = Callable[[Any, GraphQLResolveInfo, KwArg[Any]], AsyncGenerator]
Receives same arguments as Resolver.
SubscriptionResult
SubscriptionResult = Tuple[bool, Union[List[dict], AsyncGenerator[ExecutionResult, None]]]
Result type for subscribe function.
It's a tuple of two elements:
bool: True when query was executed successfully (without any errors), False otherwise.
dict or generator: JSON-serializable query result or asynchronous generator with subscription's results. Depends if query was success or not.
ValidationRules
ValidationRules = Union[Collection[Type[ASTValidationRule]], Callable[[Optional[Any], DocumentNode, dict], Optional[Collection[Type[ASTValidationRule]]]]]
Type of validation_rules option of GraphQL servers.
Enables extending of server's query validation logic.
Is either a callable that should return list of validation rules to use for GraphQL request, or list of validation rules.
Callable is evaluated with three arguments:
Optional[Any]: a context value for this request, or None.
DocumentNode: a document with parsed GraphQL query.
dict: a GraphQL request's data.