Skip to main content
Version: 0.25

ASGI handlers reference

The ariadne.asgi.handlers package exports following ASGI request handlers:

GraphQLHTTPHandler​

class GraphQLHTTPHandler(GraphQLHttpHandlerBase):
    ...

Default ASGI handler for HTTP requests.

Supports the Query and Mutation operations.

Constructor​

def __init__(
    self,
    extensions: Optional[Extensions] = None,
    middleware: Optional[Middlewares] = None,
    middleware_manager_class: Optional[Type[MiddlewareManager]] = None,
):
    ...

Initializes the HTTP handler.

Optional arguments​

extensions: an Extensions list or callable returning a list of extensions server should use during query execution. Defaults to no extensions.

middleware: a Middlewares list or callable returning a list of middlewares server should use during query execution. Defaults to no middlewares.

middleware_manager_class: a MiddlewareManager type or subclass to use for combining provided middlewares into single wrapper for resolvers by the server. Defaults to graphql.MiddlewareManager. Is only used if extensions or middleware options are set.

Methods​

handle​

async def handle(self, scope: Scope, receive: Receive, send: Send) -> None:
    ...

An entrypoint for the GraphQL HTTP handler.

This method is called by the Ariadne ASGI GraphQL application to execute queries done using the HTTP protocol.

It creates the starlette.requests.Request instance, calls handle_request method with it, then sends response back to the client.

Required arguments​

scope: The connection scope information, a dictionary that contains at least a type key specifying the protocol that is incoming.

receive: an awaitable callable that will yield a new event dictionary when one is available.

send: an awaitable callable taking a single event dictionary as a positional argument that will return once the send has been completed or the connection has been closed.

Details about the arguments and their usage are described in the ASGI specification:

https://asgi.readthedocs.io/en/latest/specs/main.html

handle_request​

async def handle_request(self, request: Request) -> Response:
    ...

Handle GraphQL request and return response for the client.

Is called by the handle method and handle_request method of the ASGI GraphQL application.

Handles three HTTP methods:

GET: returns GraphQL explorer or 405 error response if explorer or introspection is disabled.

POST: executes the GraphQL query from either application/json or multipart/form-data requests.

OPTIONS: returns supported HTTP methods.

Returns Starlette's Response instance, which is also works in FastAPI.

Required arguments:​

request: the Request instance from Starlette or FastAPI.

render_explorer​

async def render_explorer(self, request: Request, explorer: Explorer) -> Response:
    ...

Return a HTML response with GraphQL explorer.

Required arguments:​

request: the Request instance from Starlette or FastAPI.

explorer: an Explorer instance that implements the html(request: Request) method which returns either the str with HTML or None. If explorer returns None, 405 method not allowed response is returned instead.

graphql_http_server​

async def graphql_http_server(self, request: Request) -> Response:
    ...

Handles the HTTP request with GraphQL query.

Extracts GraphQL query data from requests and then executes it using the execute_graphql_query method.

Returns the JSON response from Sta

If request's data was invalid or missing, plaintext response with error message and 400 status code is returned instead.

Required arguments:​

request: the Request instance from Starlette or FastAPI.

extract_data_from_request​

async def extract_data_from_request(self, request: Request) -> Union[dict, list]:
    ...

Extracts GraphQL request data from request.

Returns a dict or list with GraphQL query data that was not yet validated.

Required arguments​

request: the Request instance from Starlette or FastAPI.

extract_data_from_json_request​

async def extract_data_from_json_request(self, request: Request) -> dict:
    ...

Extracts GraphQL data from JSON request.

Returns a dict with GraphQL query data that was not yet validated.

Required arguments​

request: the Request instance from Starlette or FastAPI.

extract_data_from_multipart_request​

async def extract_data_from_multipart_request(
    self,
    request: Request,
) -> Union[dict, list]:
    ...

Extracts GraphQL data from multipart/form-data request.

Returns an unvalidated dict or list with GraphQL query data.

Required arguments​

request: the Request instance from Starlette or FastAPI.

extract_data_from_get_request​

def extract_data_from_get_request(self, request: Request) -> dict:
    ...

Extracts GraphQL data from GET request's querystring.

Returns a dict with GraphQL query data that was not yet validated.

Required arguments​

request: the Request instance from Starlette or FastAPI.

execute_graphql_query​

async def execute_graphql_query(
    self,
    request: Any,
    data: Any,
    *,
    context_value: Any,
    query_document: Optional[DocumentNode],
) -> GraphQLResult:
    ...

Executes GraphQL query from request and returns GraphQLResult.

Creates GraphQL ContextValue, initializes extensions and middlewares, then runs the graphql function from Ariadne to execute the query.

Requires arguments​

request: the Request instance from Starlette or FastAPI.

data: a GraphQL data.

Optional arguments​

context_value: a ContextValue for this request.

query_document: an already parsed GraphQL query. Setting this option will prevent graphql from parsing query string from data second time.

get_extensions_for_request​

async def get_extensions_for_request(
    self,
    request: Any,
    context: Optional[ContextValue],
) -> ExtensionList:
    ...

Returns extensions to use when handling the GraphQL request.

Returns ExtensionList, a list of extensions to use or None.

Required arguments​

request: the Request instance from Starlette or FastAPI.

context: a ContextValue for this request.

get_middleware_for_request​

async def get_middleware_for_request(
    self,
    request: Any,
    context: Optional[ContextValue],
) -> MiddlewareList:
    ...

Returns GraphQL middlewares to use when handling the GraphQL request.

Returns MiddlewareList, a list of middlewares to use or None.

Required arguments​

request: the Request instance from Starlette or FastAPI.

context: a ContextValue for this request.

create_json_response​

async def create_json_response(
    self,
    request: Request,
    result: dict,
    success: bool,
) -> Response:
    ...

Creates JSON response from GraphQL's query result.

Returns Starlette's JSONResponse instance that's also compatible with FastAPI. If success is True, response's status code is 200. Status code 400 is used otherwise.

Required arguments​

request: the Request instance from Starlette or FastAPI.

result: a JSON-serializable dict with query result.

success: a bool specifying if

handle_not_allowed_method​

def handle_not_allowed_method(self, request: Request) -> None:
    ...

Handles request for unsupported HTTP method.

Returns 200 response for OPTIONS request and 405 response for other methods. All responses have empty body.

Required arguments​

request: the Request instance from Starlette or FastAPI.

GraphQLHandler​

class GraphQLHandler(ABC):
    ...

Base class for ASGI connection handlers.

Constructor​

def __init__(self):
    ...

Initialize the handler instance with empty configuration.

Methods​

handle​

async def handle(self, scope: Scope, receive: Receive, send: Send) -> None:
    ...

An entrypoint for the ASGI connection handler.

This method is called by Ariadne ASGI GraphQL application. Subclasses should replace it with custom implementation.

Required arguments​

scope: The connection scope information, a dictionary that contains at least a type key specifying the protocol that is incoming.

receive: an awaitable callable that will yield a new event dictionary when one is available.

send: an awaitable callable taking a single event dictionary as a positional argument that will return once the send has been completed or the connection has been closed.

Details about the arguments and their usage are described in the ASGI specification:

https://asgi.readthedocs.io/en/latest/specs/main.html

configure​

def configure(
    self,
    schema: GraphQLSchema,
    context_value: Optional[ContextValue] = None,
    root_value: Optional[RootValue] = None,
    query_parser: Optional[QueryParser] = None,
    query_validator: Optional[QueryValidator] = None,
    validation_rules: Optional[ValidationRules] = None,
    execute_get_queries: bool = False,
    debug: bool = False,
    introspection: bool = True,
    explorer: Optional[Explorer] = None,
    logger: Union[None, str, Logger, LoggerAdapter] = None,
    error_formatter: ErrorFormatter = format_error,
    execution_context_class: Optional[Type[ExecutionContext]] = None,
) -> None:
    ...

Configures the handler with options from the ASGI application.

Called by Ariadne ASGI GraphQL application as part of its initialization, propagating the configuration to it's handlers.

get_context_for_request​

async def get_context_for_request(self, request: Any, data: dict) -> Any:
    ...

Returns GraphQL context value for ASGI connection.

Resolves final context value from the ContextValue value passed to context_value option. If context_value is None, sets default context value instead, which is a dict with single request key that contains either starlette.requests.Request instance or starlette.websockets.WebSocket instance.

Required arguments​

request: an instance of ASGI connection. It's type depends on handler.

data: a GraphQL data from connection.

GraphQLHttpHandlerBase​

class GraphQLHttpHandlerBase(GraphQLHandler):
    ...

Base class for ASGI HTTP connection handlers.

Methods​

handle_request​

async def handle_request(self, request: Any) -> Any:
    ...

Abstract method for handling the request.

Should return valid ASGI response.

execute_graphql_query​

async def execute_graphql_query(
    self,
    request: Any,
    data: Any,
    *,
    context_value: Optional[Any],
    query_document: Optional[DocumentNode],
) -> GraphQLResult:
    ...

Abstract method for GraphQL query execution.

GraphQLTransportWSHandler​

class GraphQLTransportWSHandler(GraphQLWebsocketHandler):
    ...

Implementation of the (newer) graphql-transport-ws subprotocol from the graphql-ws library.

For more details see it's GH page:

https://github.com/enisdenjo/graphql-ws/blob/master/PROTOCOL.md

Constructor​

def __init__(
    self,
    *args,
    connection_init_wait_timeout: timedelta,
    **kwargs,
):
    ...

Initializes the websocket handler.

Optional arguments​

connection_init_wait_timeout: a timedelta with timeout for new websocket connections before first message is received. Defaults to 60 seconds.

Methods​

handle​

async def handle(self, scope: Scope, receive: Receive, send: Send) -> None:
    ...

An entrypoint for the GraphQL WebSocket handler.

This method is called by the Ariadne ASGI GraphQL application to handle the websocket connections.

It creates the starlette.websockets.WebSocket instance and calls handle_websocket method with it.

Required arguments​

scope: The connection scope information, a dictionary that contains at least a type key specifying the protocol that is incoming.

receive: an awaitable callable that will yield a new event dictionary when one is available.

send: an awaitable callable taking a single event dictionary as a positional argument that will return once the send has been completed or the connection has been closed.

Details about the arguments and their usage are described in the ASGI specification:

https://asgi.readthedocs.io/en/latest/specs/main.html

handle_websocket​

async def handle_websocket(self, websocket: WebSocket) -> None:
    ...

Handle GraphQL the WebSocket connection.

Is called by the handle method and handle_websocket method of the ASGI GraphQL application.

Required arguments:​

websocket: the WebSocket instance from Starlette or FastAPI.

handle_connection_init_timeout​

async def handle_connection_init_timeout(
    self,
    websocket: WebSocket,
    client_context: ClientContext,
) -> None:
    ...

handle_websocket_message​

async def handle_websocket_message(
    self,
    websocket: WebSocket,
    message: dict,
    client_context: ClientContext,
) -> None:
    ...

Handles new message from websocket connection.

Required arguments​

websocket: the WebSocket instance from Starlette or FastAPI.

message: a dict with message payload.

client_context: a ClientContext object with extra state of current websocket connection.

handle_websocket_connection_init_message​

async def handle_websocket_connection_init_message(
    self,
    websocket: WebSocket,
    message: dict,
    client_context: ClientContext,
) -> None:
    ...

Handles connection_init websocket message.

Required arguments​

websocket: the WebSocket instance from Starlette or FastAPI.

message: a dict with message payload.

client_context: a ClientContext object with extra state of current websocket connection.

handle_websocket_ping_message​

async def handle_websocket_ping_message(
    self,
    websocket: WebSocket,
    client_context: ClientContext,
) -> None:
    ...

Handles ping websocket message, answering with pong message.

Required arguments​

websocket: the WebSocket instance from Starlette or FastAPI.

client_context: a ClientContext object with extra state of current websocket connection.

handle_websocket_pong_message​

async def handle_websocket_pong_message(
    self,
    websocket: WebSocket,
    client_context: ClientContext,
) -> None:
    ...

Handles pong websocket message.

Unlike ping message, pong is unidirectional heartbeat sent by the client to the server. It doesn't require a result.

Required arguments​

websocket: the WebSocket instance from Starlette or FastAPI.

client_context: a ClientContext object with extra state of current websocket connection.

handle_websocket_complete_message​

async def handle_websocket_complete_message(
    self,
    websocket: WebSocket,
    operation_id: str,
    client_context: ClientContext,
) -> None:
    ...

Handles complete websocket message.

complete message tells the GraphQL server to stop sending events for GraphQL operation specified in the message

Required arguments​

websocket: the WebSocket instance from Starlette or FastAPI.

operation_id: a str with id of operation that should be stopped.

client_context: a ClientContext object with extra state of current websocket connection.

handle_websocket_subscribe​

async def handle_websocket_subscribe(
    self,
    websocket: WebSocket,
    data: Any,
    operation_id: str,
    client_context: ClientContext,
) -> None:
    ...

Handles subscribe websocket message.

Required arguments​

websocket: the WebSocket instance from Starlette or FastAPI.

data: any data from subscribe message.

operation_id: a str with id of new subscribe operation.

client_context: a ClientContext object with extra state of current websocket connection.

handle_websocket_invalid_type​

async def handle_websocket_invalid_type(self, websocket: WebSocket) -> None:
    ...

Handles unsupported or invalid websocket message.

Closes open websocket connection with error code 4400.

Required arguments​

websocket: the WebSocket instance from Starlette or FastAPI.

handle_on_complete​

async def handle_on_complete(
    self,
    websocket: WebSocket,
    operation: Operation,
) -> None:
    ...

Handles completed websocket operation.

Required arguments​

websocket: the WebSocket instance from Starlette or FastAPI.

operation: a completed Operation.

stop_websocket_operation​

async def stop_websocket_operation(
    self,
    websocket: WebSocket,
    operation_id: str,
    client_context: ClientContext,
) -> None:
    ...

Stops specified GraphQL operation for given connection and context.

Required arguments​

websocket: the WebSocket instance from Starlette or FastAPI.

operation_id: a str with id of operation to stop.

client_context: a ClientContext object with extra state of current websocket connection.

observe_async_results​

async def observe_async_results(
    self,
    websocket: WebSocket,
    results_producer: AsyncGenerator,
    operation_id: str,
    client_context: ClientContext,
) -> None:
    ...

Converts results from Ariadne's subscribe generator into websocket messages it next sends to the client.

Required arguments​

websocket: the WebSocket instance from Starlette or FastAPI.

results_producer: the AsyncGenerator returned from Ariadne's subscribe function.

operation_id: a str with id of operation.

client_context: a ClientContext object with extra state of current websocket connection.

GraphQLWSHandler​

class GraphQLWSHandler(GraphQLWebsocketHandler):
    ...

Implementation of the (older) graphql-ws subprotocol from the subscriptions-transport-ws library.

For more details see it's GH page:

https://github.com/apollographql/subscriptions-transport-ws/blob/master/PROTOCOL.md

Constructor​

def __init__(self, *args, keepalive: Optional[float], **kwargs):
    ...

Initializes the websocket handler.

Optional arguments​

keepalive: a float with time frequency for sending the keep-alive messages to clients with open websocket connections. 1.0 is 1 second. If set to None or 0, no keep-alive messages are sent. Defaults to None.

Methods​

handle​

async def handle(self, scope: Scope, receive: Receive, send: Send) -> None:
    ...

An entrypoint for the GraphQL WebSocket handler.

This method is called by the Ariadne ASGI GraphQL application to handle the websocket connections.

It creates the starlette.websockets.WebSocket instance and calls handle_websocket method with it.

Required arguments​

scope: The connection scope information, a dictionary that contains at least a type key specifying the protocol that is incoming.

receive: an awaitable callable that will yield a new event dictionary when one is available.

send: an awaitable callable taking a single event dictionary as a positional argument that will return once the send has been completed or the connection has been closed.

Details about the arguments and their usage are described in the ASGI specification:

https://asgi.readthedocs.io/en/latest/specs/main.html

handle_websocket​

async def handle_websocket(self, websocket: WebSocket) -> None:
    ...

Handle GraphQL the WebSocket connection.

Is called by the handle method and handle_websocket method of the ASGI GraphQL application.

Required arguments:​

websocket: the WebSocket instance from Starlette or FastAPI.

handle_websocket_message​

async def handle_websocket_message(
    self,
    websocket: WebSocket,
    message: dict,
    operations: Dict[str, Operation],
) -> None:
    ...

Handles new message from websocket connection.

Required arguments​

websocket: the WebSocket instance from Starlette or FastAPI.

message: a dict with message payload.

operations: a dict with currently active GraphQL operations.

process_single_message​

async def process_single_message(
    self,
    websocket: WebSocket,
    data: Any,
    operation_id: str,
    operations: Dict[str, Operation],
) -> None:
    ...

Processes websocket message containing new GraphQL operation.

Required arguments​

websocket: the WebSocket instance from Starlette or FastAPI.

data: a dict with message's payload.

operation_id: a str with an ID of new operation.

operations: a dict with currently active GraphQL operations.

handle_websocket_connection_init_message​

async def handle_websocket_connection_init_message(
    self,
    websocket: WebSocket,
    message: dict,
) -> None:
    ...

Handles connection_init websocket message.

Initializes new websocket instance.

Required arguments​

websocket: the WebSocket instance from Starlette or FastAPI.

message: a dict with message's payload.

handle_websocket_connection_terminate_message​

async def handle_websocket_connection_terminate_message(
    self,
    websocket: WebSocket,
) -> None:
    ...

Handles terminate websocket message.

Closes open websocket connection.

Required arguments​

websocket: the WebSocket instance from Starlette or FastAPI.

keep_websocket_alive​

async def keep_websocket_alive(self, websocket: WebSocket) -> None:
    ...

start_websocket_operation​

async def start_websocket_operation(
    self,
    websocket: WebSocket,
    data: Any,
    context_value: Any,
    query_document: DocumentNode,
    operation_id: str,
    operations: Dict[str, Operation],
) -> None:
    ...

stop_websocket_operation​

async def stop_websocket_operation(
    self,
    websocket: WebSocket,
    operation: Operation,
) -> None:
    ...

Stops specified GraphQL operation for given connection.

Required arguments​

websocket: the WebSocket instance from Starlette or FastAPI.

operation: an Operation to stop.

observe_async_results​

async def observe_async_results(
    self,
    websocket: WebSocket,
    results: AsyncGenerator,
    operation_id: str,
) -> None:
    ...

Converts results from Ariadne's subscribe generator into websocket messages it next sends to the client.

Required arguments​

websocket: the WebSocket instance from Starlette or FastAPI.

results: the AsyncGenerator returned from Ariadne's subscribe function.

operation_id: a str with id of operation.

GraphQLWebsocketHandler​

class GraphQLWebsocketHandler(GraphQLHandler):
    ...

Base class for ASGI websocket connection handlers.

Constructor​

def __init__(
    self,
    on_connect: Optional[OnConnect] = None,
    on_disconnect: Optional[OnDisconnect] = None,
    on_operation: Optional[OnOperation] = None,
    on_complete: Optional[OnComplete] = None,
):
    ...

Initialize websocket handler with optional options specific to it.

Optional arguments:​

on_connect: an OnConnect callback used on new websocket connection.

on_disconnect: an OnDisconnect callback used when existing websocket connection is closed.

on_operation: an OnOperation callback, used when new GraphQL operation is received from websocket connection.

on_complete: an OnComplete callback, used when GraphQL operation received over the websocket connection was completed.

Methods​

handle_websocket​

async def handle_websocket(self, websocket: Any) -> None:
    ...

Abstract method for handling the websocket connection.

configure​

def configure(
    self,
    *args,
    http_handler: Optional[GraphQLHttpHandlerBase],
    **kwargs,
) -> None:
    ...

Configures the handler with options from the ASGI application.

Called by Ariadne ASGI GraphQL application as part of its initialization, propagating the configuration to it's handlers.

Optional arguments​

http_handler: the GraphQLHttpHandlerBase subclass instance to use to execute the Query and Mutation operations made over the websocket connections.