An asynchronous GraphQL client built on top of aiohttp and graphql-core-next. It supports multiple transports, including aiohttp (default) and httpx. The client by default introspects schemas and validates all queries prior to dispatching to the server.
For the most recent project documentation, you can visit https://aiographql-client.readthedocs.io/.
To install the client with the default aiohttp transport:
pip install aiographql-client[aiohttp]To use httpx as the transport (includes websockets for subscriptions):
pip install aiographql-client[httpx]To install the client with the default aiohttp transport and pydantic support:
pip install aiographql-client[aiohttp,pydantic]The client supports multiple HTTP backends. You can choose which one to use during installation via extras.
aiohttp: The default asynchronous HTTP client for the library. Supports both queries and subscriptions.httpx: A modern, feature-rich HTTP client. Supports queries and subscriptions (viawebsockets).
When you initialize a GraphQLClient without specifying a transport, it will try to automatically detect an available backend:
- It first checks if
aiohttpis installed. - If not, it checks if
httpxis installed. - If neither is found, it raises a
RuntimeError.
- It first checks if
aiohttpis installed. - If not, it checks if
websocketsis installed. - If neither is found, it raises a
RuntimeError.
Note that websockets is installed automatically when the httpx extra is used.
from aiographql.client import GraphQLClient
# Auto-detects available transport (aiohttp preferred if both installed)
client = GraphQLClient(endpoint="https://api.github.com/graphql")You can also explicitly specify which transport to use by providing a transport instance:
from aiographql.client import GraphQLClient
from aiographql.client.transport import HttpxTransport, AiohttpTransport, WebsocketSubscriptionTransport
# Explicitly use httpx
transport = HttpxTransport(endpoint="https://api.github.com/graphql")
client = GraphQLClient(
endpoint="https://api.github.com/graphql",
transport=transport
)
# Explicitly use aiohttp
transport = AiohttpTransport(endpoint="https://api.github.com/graphql")
client = GraphQLClient(
endpoint="https://api.github.com/graphql",
transport=transport
)
# Explicitly use websockets for subscriptions
from aiographql.client.transport.websocket import WebsocketSubscriptionTransport
subscription_transport = WebsocketSubscriptionTransport(endpoint="wss://your-api.com/graphql")
client = GraphQLClient(
endpoint="https://your-api.com/graphql",
subscription_transport=subscription_transport
)To run the full test suite with all supported transport combinations and Python versions, you can use tox:
# Install tox
pip install tox
# Run all environments
tox
# Run a specific environment (e.g., Python 3.10 with aiohttp and pydantic)
tox -e py310-aiohttp-pydantic
# Run with only aiohttp
tox -e py310-aiohttp
# Run with only httpx
tox -e py310-httpxSee pyproject.toml for all available environment factors and combinations.
Here are some example usages of this client implementation. For more examples, and advanced scenarios, see Usage Examples section in the documentation or the examples directory in this repository.
We provide several well-documented, scenario-based examples using a Strawberry GraphQL server. You can find them in the examples/ directory.
To run the examples, first start the Strawberry server using podman compose:
podman compose run --build strawberry-serverThen you can run any of the scenario scripts:
# Basic Queries
poetry run python examples/basic_queries.py
# Pydantic Data Models
poetry run python examples/data_models.py
# Dataclass Models
poetry run python examples/dataclass_models.py
# Authenticated Requests
poetry run python examples/authenticated_requests.py
# Mutations with Variables
poetry run python examples/mutations.py
# Real-time Subscriptions
poetry run python examples/subscriptions.py
# Httpx Transport & Shared Sessions
poetry run python examples/httpx_transport.py
# Custom Serialization & Codecs
poetry run python examples/custom_serialization.pySee examples/README.md for more detailed descriptions of each scenario.
async def get_logged_in_username(token: str) -> GraphQLResponse:
client = GraphQLClient(
endpoint="https://api.github.com/graphql",
headers={"Authorization": f"Bearer {token}"},
)
request = GraphQLRequest(
query="""
query {
viewer {
login
}
}
"""
)
return await client.query(request=request)>>> import asyncio
>>> response = asyncio.run(get_logged_in_username("<TOKEN FROM GITHUB GRAPHQL API>"))
>>> response.data
{'viewer': {'login': 'username'}}async def print_city_updates(client: GraphQLClient, city: str) -> None:
request = GraphQLRequest(
query="""
subscription ($city:String!) {
city(where: {name: {_eq: $city}}) {
description
id
}
}
""",
variables={"city": city},
)
# subscribe to data and error events, and print them
await client.subscribe(
request=request, on_data=print, on_error=print, wait=True
)For custom event specific callback registration, see Callback Registry Documentation.
If your query is invalid, thanks to graphql-core-next, we get a detailed exception in the traceback.
aiographql.client.exceptions.GraphQLClientValidationException: Query validation failed
Cannot query field 'ids' on type 'chatbot'. Did you mean 'id'?
GraphQL request (4:13)
3: chatbot {
4: ids, bot_names
^
5: }
Cannot query field 'bot_names' on type 'chatbot'. Did you mean 'bot_name' or 'bot_language'?
GraphQL request (4:18)
3: chatbot {
4: ids, bot_names
^
5: }
Support for multi-operation requests and variables is available via the client. For example, the following request contains multiple operations. The instance specifies default values to use.
request = GraphQLRequest(
query="""
query get_bot_created($id: Int) {
chatbot(where: {id: {_eq: $id}}) {
id, created
}
}
query get_bot_name($id: Int) {
chatbot(where: {id: {_eq: $id}}) {
id, bot_name
}
}
""",
variables={"id": 109},
operation="get_bot_name"
)The default values can be overridden at the time of making the request if required.
await client.query(request=request, variables={"id": 20}, operation="get_bot_created")The client supports explicit decoding of results into your own models, like dataclasses or Pydantic (v2) models.
from pydantic import BaseModel
from aiographql.client import GraphQLClient
class User(BaseModel):
id: int
name: str
client = GraphQLClient(endpoint="http://localhost/graphql")
# Explicitly decode into a Pydantic model
user = await client.query_data_as(
"{ user(id: 1) { id name } }",
User,
path="user"
)
print(user.name) # AlicePass models directly into variables:
# Model as variable
user_input = CreateUserInput(name="Alice")
response = await client.query(query, variables={"input": user_input})See Data Models Documentation for more information.