Asynchronous GraphQL Client#

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.

Documentation#

For the most recent project documentation, you can visit https://aiographql-client.readthedocs.io/.

Installation#

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]

Transports & Auto-detection#

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 (via websockets).

Auto-detection#

When you initialize a GraphQLClient without specifying a transport, it will try to automatically detect an available backend:

HTTP Transport (Queries/Mutations)#

It first checks if aiohttp is installed.
If not, it checks if httpx is installed.
If neither is found, it raises a RuntimeError.

Subscription Transport#

It first checks if aiohttp is installed.
If not, it checks if websockets is 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")

Manual Selection#

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
)

Development#

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-httpx

See pyproject.toml for all available environment factors and combinations.

Example Usage#

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.

Examples#

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-server

Then 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.py

See examples/README.md for more detailed descriptions of each scenario.

Simple Query#

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'}}

Query Subscription#

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.

Query Validation Failures#

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: }

Query Variables & Operations#

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")

Typed Usage (Bring Your Own Models)#

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) # Alice

Pass 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.