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.