gql
gql is a Python GraphQL client that separates GraphQL documents and schema validation from the network transport. It can send synchronous or asynchronous queries and mutations over requests, HTTPX, or aiohttp; run subscriptions over WebSockets, multipart HTTP, Phoenix channels, or AWS AppSync; upload files; batch operations; and compose queries with a DSL. Give it a local schema or fetch one by introspection and it validates documents before the server sees them.
gql earns its weight when Python needs schemas, async sessions, subscriptions, uploads, or transport choice. For fixed HTTP queries it is more machinery than necessary, and it will not give you typed generated models or an Apollo-like cache.
Use it if
- Your Python service needs reusable async GraphQL sessions, concurrent operations, or subscriptions rather than one-off HTTP posts
- You want local query validation against a checked-in schema or an introspected server schema
- You need GraphQL multipart file uploads, WebSocket protocols, Phoenix channels, or AWS AppSync authentication and realtime support
- You need to choose among requests, HTTPX, aiohttp, and WebSockets without rewriting the document and client layers
- You only send one or two fixed GraphQL-over-HTTP requests: httpx or requests can post query and variables with less client architecture and fewer dependencies
- You expect generated typed result models: gql returns dictionaries by default and its DSL builds documents, but it is not a schema-to-Python code generator
- You need an Apollo-style normalized cache, reactive watchers, pagination policies, or client-side state management: gql is a transport and execution client, not a front-end data cache
- You must use synchronous execution inside Jupyter, ASGI, or another running asyncio loop: the README warns that the basic sync path will not work there, so the application must adopt async usage
- Your server disables introspection and you do not have a schema file: fetch_schema_from_transport will fail, leaving you with server-side validation only unless you obtain the schema separately
Setup reality
A plain pip install gql installs graphql-core, yarl, backoff, and anyio, but no concrete HTTP or WebSocket transport. Install the narrow extra you need, such as gql[requests], gql[httpx], gql[aiohttp], or gql[websockets], or use gql[all] and accept requests, requests-toolbelt, HTTPX, aiohttp, websockets, botocore, and aiofiles together. Version 4.0.0 requires Python 3.8.1 or newer. Transport choice changes the programming model. RequestsHTTPTransport and HTTPXTransport suit ordinary synchronous code; AIOHTTPTransport and HTTPXAsyncTransport support async sessions; WebsocketsTransport handles subscription protocols. The README's synchronous AIOHTTP example creates its own loop and fails when an event loop is already running, including common Jupyter and ASGI contexts. In those environments use async with Client and await the session. Opening one client context per operation repeatedly creates connections and prevents pooling, so keep a session for a unit of concurrent work, but do not reuse a closed session. Authentication is normally a transport header or transport-specific auth object; token refresh, secret storage, TLS verification, proxies, timeouts, and retry policy remain application work. fetch_schema_from_transport sends an introspection query on connection. That adds startup latency and fails against servers that disable or restrict introspection, so production services often check in a schema and update it deliberately. Local validation only happens when a schema is supplied or fetched. GraphQL transport success is not business success: a 200 response containing errors raises TransportQueryError, while malformed protocol replies, HTTP failures, TLS or connection failures, and local GraphQL validation raise different exception families. Subscriptions need a server-supported WebSocket protocol, reconnect strategy, task cancellation, and backpressure decisions. File uploads require upload_files=True and FileVar; non-streaming uploads load complete files into memory, while streaming is supported only by the aiohttp transport. Never interpolate values into a query string; use variable_values so GraphQL parsing, nullability, and input coercion remain correct.
Patterns
Execute a synchronous HTTP querysync-http-query
from gql import Client, gql
from gql.transport.requests import RequestsHTTPTransport
transport = RequestsHTTPTransport(url='https://api.example.com/graphql', timeout=10)
client = Client(transport=transport)
query = gql('query { viewer { id name } }')
result = client.execute(query)Install gql[requests]. Use the async client instead when an asyncio event loop is already running.
Reuse an asynchronous HTTP sessionasync-http-query
from gql import Client, gql
from gql.transport.aiohttp import AIOHTTPTransport
transport = AIOHTTPTransport(url='https://api.example.com/graphql')
client = Client(transport=transport)
async with client as session:
result = await session.execute(gql('query { viewer { id name } }'))Install gql[aiohttp]. The context opens and closes the transport; execute through session while it is open.
Pass typed GraphQL variablessend-variables
query = gql('''
query User($id: ID!) {
user(id: $id) { id name email }
}
''')
result = client.execute(query, variable_values={'id': user_id})Use variable_values rather than string interpolation so GraphQL input coercion and nullability rules apply.
Attach a bearer token to HTTP requestsbearer-authentication
transport = RequestsHTTPTransport(
url='https://api.example.com/graphql',
headers={'Authorization': f'Bearer {token}'},
timeout=10,
)
client = Client(transport=transport)A static header does not refresh expiring tokens. Recreate or update transport authentication through an application-owned credential flow.
Fetch and use the schema by introspectionfetch-server-schema
client = Client(
transport=transport,
fetch_schema_from_transport=True,
)
result = client.execute(gql('query { viewer { id } }'))This adds an introspection request when connecting and fails when the server blocks introspection or the credential cannot access it.
Validate queries against a checked-in schemavalidate-local-schema
from pathlib import Path
from gql import Client, gql
schema = Path('schema.graphql').read_text(encoding='utf-8')
client = Client(transport=transport, schema=schema)
result = client.execute(gql('query { viewer { id } }'))A stale local schema can reject a valid new query or accept a field the server removed. Update it through a controlled schema workflow.
Execute a mutation with input variablesexecute-mutation
mutation = gql('''
mutation RenameUser($id: ID!, $name: String!) {
renameUser(id: $id, name: $name) { id name }
}
''')
result = client.execute(
mutation,
variable_values={'id': user_id, 'name': 'Ada'},
)GraphQL mutations are not automatically idempotent. Retry only when the server operation and client request design make it safe.
Consume a subscription over WebSocketssubscribe-websocket
from gql.transport.websockets import WebsocketsTransport
transport = WebsocketsTransport(
url='wss://api.example.com/graphql',
headers={'Authorization': f'Bearer {token}'},
)
client = Client(transport=transport)
subscription = gql('subscription { jobChanged { id status } }')
async with client as session:
async for event in session.subscribe(subscription):
print(event['jobChanged'])Install gql[websockets] and confirm which GraphQL WebSocket protocol the server supports. Add application-level reconnect and cancellation logic.
Upload a file through the multipart specificationupload-file
from gql import FileVar, gql
mutation = gql('''
mutation Upload($file: Upload!) {
upload(file: $file) { id }
}
''')
mutation.variable_values = {
'file': FileVar('report.pdf', content_type='application/pdf')
}
result = client.execute(mutation, upload_files=True)Non-streaming transports load the full file into memory. Version 4 streaming uploads require AIOHTTPTransport and FileVar(streaming=True).
Run independent queries concurrentlyrun-concurrent-queries
import asyncio
async with client as session:
user_result, team_result = await asyncio.gather(
session.execute(user_query, variable_values={'id': user_id}),
session.execute(team_query, variable_values={'id': team_id}),
)Reuse one open async session. Server concurrency limits and transport connection limits still apply.
Handle GraphQL errors separately from HTTP failureshandle-query-errors
from gql.transport.exceptions import TransportQueryError, TransportServerError
try:
result = client.execute(query, variable_values=variables)
except TransportQueryError as exc:
print('GraphQL errors:', exc.errors)
raise
except TransportServerError as exc:
print('HTTP status:', exc.code)
raiseA GraphQL server can return HTTP 200 with an errors array; gql raises TransportQueryError for that response.
Download a schema for checked-in validationdownload-schema-cli
gql-cli https://api.example.com/graphql --print-schema > schema.graphqlThe endpoint must permit introspection and may require CLI header options. Review schema changes before replacing the checked-in file.
Alternatives
| Package | Registry | Pick it when |
|---|---|---|
| sgqlc | PyPI | You want schema code generation and a more typed Python GraphQL model in addition to endpoint calls |
| python-graphql-client | PyPI | You want a smaller query, mutation, and subscription client and can accept a much narrower ecosystem |
| httpx | PyPI | You only need straightforward GraphQL-over-HTTP requests and prefer to handle response errors and documents yourself |