Configuring Transport#

The client allows you to customize how GraphQL requests are executed by using a transport abstraction. By default, the client automatically selects the best available transport based on installed packages. If both aiohttp and httpx are installed, AiohttpTransport is preferred.

You can also explicitly specify a transport instance, such as aiographql.client.transport.AiohttpTransport or aiographql.client.transport.HttpxTransport.

Automatic Transport Selection#

If you do not provide a transport instance, the client will attempt to resolve a default one for you. This is equivalent to passing transport=None.

from aiographql.client import GraphQLClient

# Automatically selects the best available transport
client = GraphQLClient(endpoint="http://127.0.0.1:8080/v1/graphql")

Available Transports#

The library supports multiple transports for HTTP (Queries/Mutations) and WebSocket (Subscriptions).

Subscription Lifecycle#

Subscriptions ride a long-lived WebSocket connection following the GraphQL over WebSocket protocol. The client negotiates the modern graphql-transport-ws protocol by default and automatically falls back to the legacy graphql-ws protocol for older servers. The sequence below shows the handshake and message flow for a single subscription:

sequenceDiagram

autonumber participant App as Application participant Client as GraphQLClient participant WS as Subscription Transport participant Server as GraphQL Server

App->>Client: subscribe(query) Client->>WS: connect() WS->>Server: WebSocket upgrade (protocols negotiation) Server-->>WS: connection_ack WS->>Server: subscribe/start { id, query } loop while subscription active Server-->>WS: next/data { id, payload } WS-->>Client: yield CallbackRegistry events Client-->>App: GraphQLResponse end App->>Client: unsubscribe() Client->>WS: complete/stop { id } WS->>Server: complete/stop { id } Server-->>WS: complete { id } WS-->>Client: close()

Protocol Negotiation#

The client defaults to negotiating both graphql-transport-ws (modern) and graphql-ws (legacy) protocols during the WebSocket handshake. The server selects which protocol to use, and the client adapts its message format automatically:

• Modern protocol (graphql-transport-ws): Uses subscribe, next, complete, ping, and pong messages.

• Legacy protocol (graphql-ws): Uses start, data, stop, and ka (keep-alive) messages.

You can explicitly specify protocols using the protocols parameter:

# Force modern protocol only
await client.subscribe(query, protocols="graphql-transport-ws")

# Force legacy protocol only
await client.subscribe(query, protocols="graphql-ws")

# Default: try both (modern first, legacy fallback)
await client.subscribe(query)

Customizing Transport#

You can provide a custom transport instance when creating the aiographql.client.GraphQLClient.

from aiographql.client import GraphQLClient
from aiographql.client.transport import AiohttpTransport

transport = AiohttpTransport(endpoint="http://127.0.0.1:8080/v1/graphql")
client = GraphQLClient(endpoint="http://127.0.0.1:8080/v1/graphql", transport=transport)

Using HttpxTransport#

To use the httpx library for making requests, you can use aiographql.client.transport.HttpxTransport. This requires httpx to be installed.

from aiographql.client import GraphQLClient
from aiographql.client.transport import HttpxTransport

transport = HttpxTransport(endpoint="http://127.0.0.1:8080/v1/graphql")
client = GraphQLClient(endpoint="http://127.0.0.1:8080/v1/graphql", transport=transport)

Custom HTTP Client Sessions#

The aiographql.client.transport.AiohttpTransport allows you to specify an aiohttp Client Session for use at various levels, including per query and for all queries made by the client.

Similarly, aiographql.client.transport.HttpxTransport allows you to specify an httpx.AsyncClient.

Providing your own session is highly recommended for production use, as it allows for better control over connection pooling, timeouts, and lifecycle management.

Usage Patterns#

1. Global Session (Recommended for high performance)

   Creating a client with a pre-configured session:

import aiohttp
from aiographql.client import GraphQLClient

async with aiohttp.ClientSession(
    connector=aiohttp.TCPConnector(limit=200, force_close=True)
) as session:
    client = GraphQLClient(
        endpoint="http://127.0.0.1:8080/v1/graphql",
        session=session
    )
    await client.query("{ city { name } }")

2. Per-query Session

# Using Httpx
async with httpx.AsyncClient() as async_client:
    await client.query(
        request=request,
        session=async_client
    )

Connection Limits & Subscriptions#

By default, the AiohttpTransport uses a TCPConnector with a connection limit of 100.

Impact on Subscriptions#

GraphQL subscriptions maintain a long-lived WebSocket connection. If you are using many concurrent subscriptions, you may reach the default connection limit.

• Exhausting limits: If the limit is reached, subsequent requests (queries/mutations) may hang or time out until a connection is freed.

• Best Practice: For production applications with high subscription counts or high concurrency, provide a custom session with an appropriate limit:

import aiohttp

# Increase limit for high-concurrency or many subscriptions
connector = aiohttp.TCPConnector(limit=500)
async with aiohttp.ClientSession(connector=connector) as session:
    client = GraphQLClient(endpoint="...", session=session)
    # Use client...

Reliable Production Use#

For reliable production use, consider the following guides:

Timeouts#

Always set reasonable timeouts for your requests to prevent your application from hanging indefinitely.

import aiohttp

timeout = aiohttp.ClientTimeout(total=30, connect=10)
async with aiohttp.ClientSession(timeout=timeout) as session:
    client = GraphQLClient(endpoint="...", session=session)
    # ...

Retries#

The library does not implement automatic retries. For production, consider using a library like tenacity to handle transient network errors or rate limits.

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
async def query_with_retry(client, request):
    return await client.query(request)

Behind SOCKS Proxies#

In order use via a socks proxy, you will need to custom connector, like the one provided by aiohttp-socks.

Here is an example code snippet using this library.

connector = aiohttp_socks.ProxyConnector(
    proxy_type=aiohttp_socks.ProxyType.SOCKS5,
    host="127.0.0.1",
    port=1080,
    rdns=True,
)
async with aiohttp.ClientSession(connector=connector) as session:
    client = GraphQLClient(
        endpoint="http://gql.example.com/v1/graphql", session=session
    )
    await client.query(request="query { city { name } }")