Async GraphQL Client
Usage Examples¶
Hint
Many of the following examples make use of GitHub GraphQL API. You can retrieve a token as detailed here.
Queries¶
Simple Query¶
The aiographql.client.GraphQLClient can be used to store headers like Authorization that need to be sent with
every request made.
client = GraphQLClient(
endpoint="https://api.github.com/graphql",
headers={"Authorization": f"Bearer {TOKEN}"},
)
request = GraphQLRequest(
query="""
query {
viewer {
login
}
}
"""
)
response = await client.query(request=request)
If you have a valid token, the above query will return the following data.
>>> response.data
{'viewer': {'login': 'username'}}
If you intend to only make use of the query once, ie. it is not re-used, you may forgo
the creation of an aiographql.client.GraphQLRequest instance and pass in the
query string direct to aiographql.client.GraphQLClient.query().
await client.query(request="{ viewer { login } }")
Query Variables¶
request = GraphQLRequest(
query="""
query($number_of_repos:Int!) {
viewer {
repositories(last: $number_of_repos) {
nodes {
name
isFork
}
}
}
}
""",
variables={"number_of_repos": 3},
)
response: GraphQLResponse = await client.query(request=request)
You can override default values specified in the prepared request too. The values are upserted into the existing defaults.
response: GraphQLResponse = await client.query(request=request, variables={
"number_of_repos": 1
})
Specifying Operation Name¶
You can use a single aiographql.client.GraphQLRequest object to stop a query
wit multiple operations.
request = GraphQLRequest(
query="""
query FindFirstIssue {
repository(owner:"octocat", name:"Hello-World") {
issues(first:1) {
nodes {
id
url
}
}
}
}
query FindLastIssue {
repository(owner:"octocat", name:"Hello-World") {
issues(last:1) {
nodes {
id
url
}
}
}
}
""",
operation="FindFirstIssue",
)
# use the default operation (FindFirstIssue)
response = await client.query(request=request)
# use the operation FindLastIssue
response = await client.query(
request=request,
operation="FindLastIssue"
)
Mutations¶
Mutations are performed using the same aiographql.client.GraphQLClient.query() method, as the GraphQL specification treats them similarly to queries but with side effects.
request = GraphQLRequest(
query="""
mutation AddStar($starrableId: ID!) {
addStar(input: {starrableId: $starrableId}) {
starrable {
stargazerCount
}
}
}
""",
variables={"starrableId": "R_kgDOJ-..."}
)
response = await client.query(request)
Alternatively, you can use the aiographql.client.GraphQLClient.post() method to explicitly use a POST request (which is the default for query anyway).
response = await client.post(request)
Custom Headers per Request¶
You can provide custom headers for a specific request. These will be merged with the headers configured on the client.
response = await client.query(
request="{ viewer { login } }",
headers={"X-Custom-Header": "value"}
)
Handling Errors¶
Always check for errors in the response. The aiographql.client.GraphQLResponse object contains an errors attribute which is a list of errors returned by the server.
response = await client.query("{ viewer { nonExistentField } }")
if response.errors:
for error in response.errors:
print(f"Error: {error['message']}")
# Access extensions, locations, path if available
print(f"Code: {error.get('extensions', {}).get('code')}")
else:
print(response.data)
Subscriptions¶
Subscriptions¶
The following example makes use of the Hasura World Database Demo application as there aren’t many public GraphQL schema that allow subscriptions for testing. You can use the project’s provided docker compose file to start an instance locally.
- By default the subscription is closed if any of the following event type is received.
The following example will subscribe to any change events and print the event as is to
stdout when either aiographql.client.GraphQLSubscriptionEventType.DATA or
aiographql.client.GraphQLSubscriptionEventType.ERROR is received.
request = GraphQLRequest(
query="""
subscription {
city(where: {name: {_eq: "Berlin"}}) {
name
id
}
}
"""
)
# subscribe to data and error events, and print them
subscription = await client.subscribe(
request=request, on_data=print, on_error=print
)
# unsubscribe
await subscription.unsubscribe_and_wait()
Hint
In the case you want to specify the GraphQL over WebSocket sub-protocol to use,
you may do so by setting aiographql.client.GraphQLSubscription.protocols.
For example, await client.subscribe(..., protocols="graphql-ws"). This is
required for certain server implementations like Apollo Server
as it supports multiple implementations.
Similarly, if your server requires additional fields in the connection_init payload (for example, an authToken or headers), you can provide them using the connection_init_payload parameter.
subscription = await client.subscribe(
request=request,
connection_init_payload={
"authToken": "my-secret-token",
"headers": {"X-Custom-Auth": "value"}
}
)
Note that headers provided in the GraphQLRequest will be merged with and take precedence over headers in connection_init_payload.
Callback Registry¶
Subscriptions make use of cafeteria.asyncio.callbacks.CallbackRegistry internally to
trigger registered callbacks when an event of a particular type is encountered. You can
also register a Coroutine if required.
# both the following statements have the same effect
subscription = await client.subscribe(
request=request, on_data=print, on_error=print
)
subscription = await client.subscribe(
request=request, callbacks={
GraphQLSubscriptionEventType.DATA: print,
GraphQLSubscriptionEventType.ERROR: print,
}
)
# this can also be done as below
registry = CallbackRegistry()
registry.register(GraphQLSubscriptionEventType.DATA, print)
registry.register(GraphQLSubscriptionEventType.ERROR, print)
If you’d like a single callback for all event types or any “unregistered” event, you can simply set the event type to None when registering the callback.
>>> registry.register(None, print)
Here is an example that will print the timestamp every time a keep-alive event is received.
subscription.callbacks.register(
GraphQLSubscriptionEventType.KEEP_ALIVE,
lambda x: print(f"Received keep-alive at {datetime.utcnow().isoformat()}")
)
Waiting for Subscriptions¶
By default, aiographql.client.GraphQLClient.subscribe() returns a aiographql.client.subscription.GraphQLSubscription object immediately, while the subscription runs in the background.
If you want to wait for the subscription to complete (e.g., in a simple script), you can pass wait=True.
await client.subscribe(request=request, on_data=print, wait=True)
Alternatively, you can wait for the background task manually:
subscription = await client.subscribe(request=request, on_data=print)
# Do other things...
await subscription.task
Error Handling in Subscriptions¶
When a subscription encounters an error, the on_error callback is triggered. It’s important to note that by default, most errors will cause the subscription to close.
async def handle_error(error):
print(f"Subscription error: {error}")
# Logic to potentially restart subscription if needed
subscription = await client.subscribe(
request=request,
on_data=print,
on_error=handle_error
)
Connection Pool Limits¶
As mentioned in the Configuring Transport section, the default connection limit is 100. Each active subscription consumes one connection from this pool. If you anticipate having many concurrent subscriptions, ensure your session’s connector is configured with a higher limit.
Validation¶
Client Side Query Validation¶
Because we make use of GraphQL Core 3, client-side query validation is performed by default.
request = GraphQLRequest(
query="""
query {
bad {
login
}
}
"""
)
response: GraphQLResponse = await client.query(request=request)
This will raise aiographql.client.GraphQLClientValidationException. The message
will also contain information about the error.
aiographql.client.exceptions.GraphQLClientValidationException: Query validation failed
Cannot query field 'bad' on type 'Query'.
GraphQL request:3:15
2 | query {
3 | bad {
| ^
4 | login
Process finished with exit code 1
Server Side Query Validation¶
You can skip the client side validation, forcing server side validation instead by setting
the aiographql.client.GraphQLRequest.validate to False before making the request.
request = GraphQLRequest(
query="""
query {
bad {
login
}
}
""",
validate=False
)
response: GraphQLResponse = await client.query(request=request)
If you are working with a GraphQL server that has introspection disabled, you can disable
validation globally when initializing the aiographql.client.GraphQLClient.
client = GraphQLClient(
endpoint="https://api.github.com/graphql",
validate=False
)
response: GraphQLResponse = await client.query("query { viewer { login } }")
>>> response.data
{}
>>> response.errors
[GraphQLError(extensions={'code': 'undefinedField', 'typeName': 'Query', 'fieldName': 'bad'}, locations=[{'line': 3, 'column': 15}], message="Field 'bad' doesn't exist on type 'Query'", path=['query', 'bad'])]
Introspection Caching¶
The client introspects the server to get the schema for validation. This schema is cached internally. For long-running production services, if the server schema changes, you can force a refresh:
await client.get_schema(refresh=True)
This is particularly useful if your GraphQL server is updated with new fields or types without restarting the client application.
Data Models¶
Bring Your Own Models¶
The aiographql-client library allows you to use your own Python models, such as dataclasses or Pydantic models, for both input variables and output decoding. This “bring your own models” workflow is explicit—the client does not automatically decode responses into models unless you request it.
Raw usage and typed usage are kept separate, giving you full control over when and how you use your models.
Supported Model Types¶
The client includes a default codec that supports:
Standard Python Dataclasses: Both for input and output.
Pydantic Models: Support for Pydantic v2 (Pydantic >= 2.0 is required).
Externally Generated Models: Any models generated from tools like
datamodel-code-generator(which typically generates Pydantic models) work seamlessly.
Dataclasses¶
You can pass a dataclass instance directly as a GraphQL variable. When receiving data, you can use query_data_as to decode the result into a dataclass.
from dataclasses import dataclass
from aiographql.client import GraphQLClient
@dataclass
class CreateUserInput:
name: str
@dataclass
class User:
id: int
name: str
client = GraphQLClient(endpoint="http://localhost/graphql")
# Input: Passing a dataclass in variables
user_input = CreateUserInput(name="Alice")
query = "mutation CreateUser($input: CreateUserInput) { createUser(input: $input) { id name } }"
# Output: Explicitly decoding into a dataclass
# The 'path' argument specifies where in the response JSON to find the data
user = await client.query_data_as(
query,
User,
variables={"input": user_input},
path="createUser"
)
print(user.id, user.name)
# 1 Alice
Pydantic Models¶
Pydantic models are supported for both input and output. The client will automatically detect if Pydantic is installed and use its validation and serialization features.
from pydantic import BaseModel
from aiographql.client import GraphQLClient
class User(BaseModel):
id: int
name: str
client = GraphQLClient(endpoint="http://localhost/graphql")
query = "{ user(id: 1) { id name } }"
# Explicitly decoding into a Pydantic model
user = await client.query_data_as(query, User, path="user")
print(user.id, user.name)
# 1 Alice
Externally Generated Models¶
If you use tools like datamodel-code-generator to generate Python models from a GraphQL schema or JSON response, you can use those generated models directly with the client’s query_data_as method.
Because these tools typically generate Pydantic models or dataclasses, they are fully compatible with the client’s built-in codec.
Note
The aiographql-client does not include built-in schema code generation. You should use external tools to generate your models and then use them with the client.
Explicit vs Automatic Decoding¶
It is important to note that decoding is explicit.
client.query(...)returns aGraphQLResponseobject whereresponse.datais a raw dictionary.client.query_data_as(..., model_class, path="...")performs the query and then explicitly decodes the data at the specified path into the providedmodel_class.
This separation ensures that you can always access the raw response if needed.