Client¶

GraphQLClient¶

class aiographql.client.GraphQLClient(endpoint: str, headers: Mapping[str, str] | None = None, method: str | None = None, schema: graphql.GraphQLSchema | None = None, session: GraphQLSession | None = None, validate: bool = True, serializer: GraphQLSerializer | None = None, codec: GraphQLCodec | None = None, transport: GraphQLTransport | None = None, subscription_transport: GraphQLSubscriptionTransport | None = None, schema_ttl: int | float | None = None) → None¶

Client implementation handling all interactions with a specified endpoint. The following example shows how to make a simple query.

client = GraphQLClient(
    endpoint="http://127.0.0.1:8080/v1/graphql",
    headers={"Authorization": "Bearer <token>"},
)
response: GraphQLResponse = await client.query("{ city { name } }")

You can also use an application scoped GraphQLSession throughout the life of the client as show below.

async with httpx.AsyncClient() as session:
    client = GraphQLClient(
        endpoint="http://127.0.0.1:8080/v1/graphql",
        session=session
    )

Parameters:

endpoint (str) – URI of graph api.
headers (Mapping[str, str] | None) – Default headers to use for every request made by this client. By default the client adds ‘Content-Type: application/json’ and ‘Accept-Encoding: gzip’ to all requests. These can be overridden by specifying then here.
method (str | None) – Default method to use when submitting a GraphQL request to the specified endpoint.
session (GraphQLSession | None) – Optional GraphQLSession to use when making requests. This is expected to be externally managed.
validate (bool) – If set to False, the client will not attempt to validate requests against the schema from the server. This is useful when introspection is disabled on the server.
serializer (GraphQLSerializer | None) – Custom JSON serializer to use for requests and responses.
codec (GraphQLCodec | None) – Custom codec to use for encoding request variables and decoding response data.
transport (GraphQLTransport | None) – Custom transport to use for making requests. If not provided, the best available transport is automatically selected (preferring aiohttp). Must be a GraphQLTransport instance.
subscription_transport (GraphQLSubscriptionTransport | None) – Custom transport to use for subscriptions. If not provided, the best available transport is automatically selected. Must be a GraphQLSubscriptionTransport instance.
schema (graphql.GraphQLSchema | None)
schema_ttl (int | float | None)

async close() → None¶

Close the underlying transport.

Return type:: None

async introspect(headers: dict[str, str] | None = None) → GraphQLSchema¶

Introspect the GraphQL endpoint specified for this client and return a graphql.GraphQLSchema object specifying the schema associated with this endpoint.

Return type:: GraphQLSchema
Returns:: GraphQL schema for the configured endpoint
Parameters:: headers (dict[str, str] | None)

async get_schema(refresh: bool = False, headers: dict[str, str] | None = None) → GraphQLSchema¶

Get the introspected schema for the endpoint used by this client. If an unexpired cache exists, this is returned unless the refresh parameter is set to True.

Parameters:

refresh (bool) – Refresh the cached schema by forcing an introspection of the GraphQL endpoint.
headers (dict[str, str] | None) – Request headers

Return type:

GraphQLSchema

Returns:

The GraphQL schema as introspected. This maybe a previously cached value.

async validate(request: GraphQLRequest, schema: GraphQLSchema | None = None, headers: dict[str, str] | None = None, force: bool = False) → None¶

Validate a given request against a schema (provided or fetched). Validation is skipped if the request’s validate property is set to False unless forced.

Parameters:

request (GraphQLRequest) – Request that is to be validated.
schema (GraphQLSchema | None) – Schema against which provided request should be validated, if different from GraphQLRequest.schema or as fetched from the client endpoint.
headers (dict[str, str] | None) – Headers to be set when fetching the schema from the client endpoint. If provided, request headers are ignored.
force (bool) – Force validation even if the provided request has validation disabled.

Return type:

None

Execute a query and decode the response data into a Python object of the specified type.

Parameters:

request (GraphQLRequest | str) – Request to send to the GraphQL server.
result_type (type[T]) – The type to decode the data into.
path (str | None) – An optional dot-separated path to the data to decode.
method (str | None) – HTTP method to use when submitting request (POST/GET).
headers (dict[str, str] | None) – Additional headers to be set when sending HTTP request.
operation (str | None) – GraphQL operation name to use.
variables (dict[str, Any] | None) – Query variables to set for the provided request.
session (GraphQLSession | None) – Optional GraphQLSession to use for requests.

Return type:

Returns:

The decoded data.

Method to send provided GraphQLRequest to the configured endpoint as an HTTP request. This method handles the configuration of headers HTTP method specific handling of request parameters and/or data as required.

The order of precedence, least to greatest, of headers is as follows,

client headers (GraphQLClient.headers)
request headers (GraphQLRequest.headers)
headers specified as method parameter

In accordance to the GraphQL specification, any non 2XX response is treated as an error and raises GraphQLTransactionException instance.

Parameters:

request (GraphQLRequest | str) – Request to send to the GraphQL server.
method (str | None) – HTTP method to use when submitting request (POST/GET). If once is not specified, the client default (GraphQLClient.method) is used.
headers (dict[str, str] | None) – Additional headers to be set when sending HTTP request.
operation (str | None) – GraphQL operation name to use if the GraphQLRequest.query contains named operations. This will override any default operation set.
variables (dict[str, Any] | None) – Query variables to set for the provided request. This will override the default values for any existing variables in the request if set.
session (GraphQLSession | None) – Optional GraphQLSession to use for requests

Return type:

GraphQLResponse

Returns:

The resulting response object.

async post(request: GraphQLRequest, headers: dict[str, str] | None = None, operation: str | None = None, variables: dict[str, Any] | None = None, session: GraphQLSession | None = None) → GraphQLResponse¶

Helper method that wraps GraphQLClient.query with method explicitly set as GraphQLQueryMethod.post.

Parameters:

request (GraphQLRequest) – Request to send to the GraphQL server.
headers (dict[str, str] | None) – Additional headers to be set when sending HTTP request.
operation (str | None) – GraphQL operation name to use if the GraphQLRequest.query contains named operations. This will override any default operation set.
variables (dict[str, Any] | None) – Query variables to set for the provided request. This will override the default values for any existing variables in the request if set.
session (GraphQLSession | None) – Optional GraphQLSession to use for requests

Return type:

GraphQLResponse

Returns:

The resulting GraphQLResponse object.

async get(request: GraphQLRequest, headers: dict[str, str] | None = None, operation: str | None = None, variables: dict[str, Any] | None = None, session: GraphQLSession | None = None) → GraphQLResponse¶

Helper method that wraps :method: GraphQLClient.query with method explicitly set as GraphQLQueryMethod.get.

Parameters:

request (GraphQLRequest) – Request to send to the GraphQL server.
headers (dict[str, str] | None) – Additional headers to be set when sending HTTP request.
operation (str | None) – GraphQL operation name to use if the GraphQLRequest.query contains named operations. This will override any default operation set.
variables (dict[str, Any] | None) – Query variables to set for the provided request. This will override the default values for any existing variables in the request if set.
session (GraphQLSession | None) – Optional GraphQLSession to use for requests

Return type:

GraphQLResponse

Returns:

The resulting GraphQLResponse object.

async subscribe(request: GraphQLRequest, headers: dict[str, str] | None = None, operation: str | None = None, variables: dict[str, Any] | None = None, callbacks: CallbacksType | None = None, on_data: CallbackType | None = None, on_error: CallbackType | None = None, session: GraphQLSession | None = None, wait: bool = False, protocols: str | Iterable[str] = ('graphql-ws',), connection_init_payload: dict[str, Any] | None = None, transport: GraphQLSubscriptionTransport | None = None) → GraphQLSubscription¶

Create and initialise a GraphQL subscription. Once subscribed and a known event is received, all registered callbacks for the event type is triggered with the aiographql.client.GraphQLSubscriptionEvent instance passed in the first argument.

The following example will start a subscription that prints all data events as it receives them.

# initialise and subscribe to events in the background
subscription: GraphQLSubscription = await client.subscribe(
    request="{ notifications: { id, summary } }",
    on_data=lambda event: print(f"Data: {event}"),
    on_error=lambda event: print(f"Error: {event}"),
)
# process events for 10 seconds then unsubscribe
await asyncio.wait(subscription.task, timeout=10)
subscription.unsubscribe()

Parameters:

request (GraphQLRequest) – Request to send to the GraphQL server.
headers (dict[str, str] | None) – Additional headers to be set when sending HTTP request.
operation (str | None) – GraphQL operation name to use if the GraphQLRequest.query contains named operations. This will override any default operation set.
variables (dict[str, Any] | None) – Query variables to set for the provided request. This will override the default values for any existing variables in the request if set.
session (GraphQLSession | None) – Optional GraphQLSession to use for requests
callbacks (CallbacksType | None) – Custom callback registry mapping an event to one more more callback methods. If not provided, a new instance is created.
on_data (CallbackType | None) – Callback to use when data event is received.
on_error (CallbackType | None) – Callback to use when an error occurs.
session – Optional session to use for connecting the graphql endpoint, if one is not provided, a new session is created for the duration of the subscription.
wait (bool) – If set to True, this method will wait until the subscription is completed, websocket disconnected or async task cancelled.
protocols (str | Iterable[str]) – GraphQL over WebSocket Sub-protocol(s) used.
connection_init_payload (dict[str, Any] | None) – Extra fields for the connection_init payload.
transport (GraphQLSubscriptionTransport | None) – Custom transport to use for the subscription. If not provided, the best available transport is automatically selected. Must be a GraphQLSubscriptionTransport instance.

Return type:

GraphQLSubscription

Returns:

The resulting GraphQLResponse object.

Returns:

The initialised subscription.

GraphQLSubscription¶

class aiographql.client.GraphQLSubscription(request: GraphQLRequest | str, headers: dataclasses.InitVar[dict[str, str] | None]=None, operation: dataclasses.InitVar[str | None] = None, variables: dataclasses.InitVar[dict[str, Any] | None]=None, codec: dataclasses.InitVar[GraphQLCodec | None] = None, callbacks: CallbacksType | None = <factory>, stop_event_types: list[GraphQLSubscriptionEventType] = <factory>, protocols: str | Iterable[str] = <factory>, connection_init_payload: dict[str, Any] | None=None, serializer: GraphQLSerializer | None = None, transport: GraphQLSubscriptionTransport | None = None) → None¶

Subscription container, with an attached cafeteria.asyncio.callbacks.CallbackRegistry. When subscribed, the task will be populated with the asyncio.Task instance.

By default the subscription will be stopped, if an error, connection error or complete (GraphQLSubscriptionEventType) is received.

Subscription instances are intended to be used as immutable objects. However, callbacks and stop_event_types can be updated after initialisation.

Parameters:

id – A unique subscription identifier that will be passed into any events generated by this subscription.
callbacks (CallbacksType | None) – A CallbackRegistry containing a mapping of GraphQLSubscriptionEventType callback methods to trigger.
stop_event_types (list[GraphQLSubscriptionEventType]) – Events that cause the subscription to stop. By default, connection error, query error or connection complete events received are considered stop events.
protocols (str | Iterable[str]) – GraphQL over WebSocket Sub-protocol used. This is optional, as this is only required for certain servers. When specified, this value is set as the Sec-WebSocket-Protocol header when a WebSocket connection is established.
connection_init_payload (dict[str, Any] | None) – Extra fields for the connection_init payload.
request (GraphQLRequest | str)
headers (dataclasses.InitVar[dict[str, str] | None])
operation (dataclasses.InitVar[str | None])
variables (dataclasses.InitVar[dict[str, Any] | None])
codec (dataclasses.InitVar[GraphQLCodec | None])
serializer (GraphQLSerializer | None)
transport (GraphQLSubscriptionTransport | None)

active() → bool¶

Check if the subscription is active.

Return type:: bool
Returns:: True if subscribed and active.

connection_init_request() → dict[str, Any]¶

Connection init payload to use when initiating a new subscription.

Return type:: dict[str, Any]
Returns:: Connection initialise payload.

connection_start_request() → dict[str, Any]¶

Connection start payload to use when starting a subscription.

Return type:: dict[str, Any]
Returns:: Connection start payload.

connection_stop_request() → dict[str, Any]¶

Connection stop payload to use when stopping a subscription.

Return type:: dict[str, Any]
Returns:: Connection stop payload.

is_stop_event(event: GraphQLSubscriptionEvent) → bool¶

Check if the provided event is configured as a stop even for this subscription.

Parameters:: event (GraphQLSubscriptionEvent) – Event to check.
Return type:: bool
Returns:: True if event is in stop_event_types.

async handle(event: GraphQLSubscriptionEvent) → None¶

Helper method to dispatch any configured callbacks for the specified event type.

Parameters:: event (GraphQLSubscriptionEvent) – Event to dispatch callbacks for.
Return type:: None

async subscribe(endpoint: str, force: bool = False, transport: GraphQLSubscriptionTransport | None = None, session: GraphQLSession | None = None, wait: bool = False) → None¶

Create a websocket subscription and set internal task.

Parameters:

endpoint (str) – GraphQL endpoint to subscribe to
force (bool) – Force re-subscription if already subscribed
transport (GraphQLSubscriptionTransport | None) – Optional transport to use for requests
session (GraphQLSession | None) – Optional session to use for requests
wait (bool) – If set to True, this method will wait until the subscription is completed, websocket disconnected or async task cancelled.

Return type:

None

unsubscribe() → None¶

Unsubscribe current websocket subscription if active and clear internal task.

Return type:: None