This is a GraphQL client for Python.
Plays nicely with graphene, graphql-core, graphql-js and any other GraphQL implementation compatible with the spec.
GQL architecture is inspired by React-Relay and Apollo-Client.
$ pip install gql
Please note that the following documentation describes the current version which is currently only available as a pre-relase and needs to be installed with
$ pip install --pre gql
The example below shows how you can execute queries against a local schema.
from gql import gql, Client
from .someSchema import SampleSchema
client = Client(schema=SampleSchema)
query = gql('''
{
hello
}
''')
client.execute(query)If you want to add additional headers when executing the query, you can specify these in a transport object:
from gql import Client
from gql.transport.requests import RequestsHTTPTransport
from .someSchema import SampleSchema
client = Client(transport=RequestsHTTPTransport(
url='/graphql', headers={'Authorization': 'token'}), schema=SampleSchema)To execute against a graphQL API. (We get the schema by using introspection).
from gql import gql, Client
from gql.transport.requests import RequestsHTTPTransport
sample_transport=RequestsHTTPTransport(
url='https://countries.trevorblades.com/',
verify=False,
retries=3,
)
client = Client(
transport=sample_transport,
fetch_schema_from_transport=True,
)
query = gql('''
query getContinents {
continents {
code
name
}
}
''')
client.execute(query)If you have a local schema stored as a schema.graphql file, you can do:
from graphql import build_ast_schema, parse
from gql import gql, Client
with open('path/to/schema.graphql') as source:
document = parse(source.read())
schema = build_ast_schema(document)
client = Client(schema=schema)
query = gql('''
{
hello
}
''')
client.execute(query)With a python version > 3.6, it is possible to execute GraphQL subscriptions using the websockets transport:
from gql import gql, Client
from gql.transport.websockets import WebsocketsTransport
sample_transport = WebsocketsTransport(url='wss:https://your_server/graphql')
client = Client(
transport=sample_transport,
fetch_schema_from_transport=True,
)
query = gql('''
subscription yourSubscription {
...
}
''')
for result in client.subscribe(query):
print (f"result = {result!s}")Note: the websockets transport can also execute queries or mutations
When using the execute or subscribe function directly on the client, the execution is synchronous.
It means that we are blocked until we receive an answer from the server and
we cannot do anything else while waiting for this answer.
It is also possible to use this library asynchronously using asyncio.
Async Features:
- Execute GraphQL subscriptions (See using the websockets transport)
- Execute GraphQL queries, mutations and subscriptions in parallel
To use the async features, you need to use an async transport:
- AIOHTTPTransport for the HTTP(s) protocols
- WebsocketsTransport for the ws(s) protocols
This transport uses the aiohttp library
GraphQL subscriptions are not supported on the HTTP transport. For subscriptions you should use the websockets transport.
from gql import gql, Client
from gql.transport.aiohttp import AIOHTTPTransport
import asyncio
async def main():
sample_transport = AIOHTTPTransport(
url='https://countries.trevorblades.com/graphql',
headers={'Authorization': 'token'}
)
async with Client(
transport=sample_transport,
fetch_schema_from_transport=True,
) as session:
# Execute single query
query = gql('''
query getContinents {
continents {
code
name
}
}
''')
result = await session.execute(query)
print(result)
asyncio.run(main())The websockets transport uses the apollo protocol described here:
Apollo websockets transport protocol
This transport allows to do multiple queries, mutations and subscriptions on the same websocket connection
import logging
logging.basicConfig(level=logging.INFO)
from gql import gql, Client
from gql.transport.websockets import WebsocketsTransport
import asyncio
async def main():
sample_transport = WebsocketsTransport(
url='wss:https://countries.trevorblades.com/graphql',
headers={'Authorization': 'token'}
)
async with Client(
transport=sample_transport,
fetch_schema_from_transport=True,
) as session:
# Execute single query
query = gql('''
query getContinents {
continents {
code
name
}
}
''')
result = await session.execute(query)
print(result)
# Request subscription
subscription = gql('''
subscription {
somethingChanged {
id
}
}
''')
async for result in session.subscribe(subscription):
print(result)
asyncio.run(main())If you need to connect to an ssl encrypted endpoint:
- use wss instead of ws in the url of the transport
- set the parameter ssl to True
import ssl
sample_transport = WebsocketsTransport(
url='wss:https://SERVER_URL:SERVER_PORT/graphql',
headers={'Authorization': 'token'}
)If you have a self-signed ssl certificate, you need to provide an ssl_context with the server public certificate:
import pathlib
import ssl
ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
localhost_pem = pathlib.Path(__file__).with_name("YOUR_SERVER_PUBLIC_CERTIFICATE.pem")
ssl_context.load_verify_locations(localhost_pem)
sample_transport = WebsocketsTransport(
url='wss:https://SERVER_URL:SERVER_PORT/graphql',
ssl=ssl_context
)If you have also need to have a client ssl certificate, add:
ssl_context.load_cert_chain(certfile='YOUR_CLIENT_CERTIFICATE.pem', keyfile='YOUR_CLIENT_CERTIFICATE_KEY.key')There are two ways to send authentication tokens with websockets depending on the server configuration.
- Using HTTP Headers
sample_transport = WebsocketsTransport(
url='wss:https://SERVER_URL:SERVER_PORT/graphql',
headers={'Authorization': 'token'}
)- With a payload in the connection_init websocket message
sample_transport = WebsocketsTransport(
url='wss:https://SERVER_URL:SERVER_PORT/graphql',
init_payload={'Authorization': 'token'}
)It is possible to send multiple GraphQL queries (query, mutation or subscription) in parallel, on the same websocket connection, using asyncio tasks
async def execute_query1():
result = await session.execute(query1)
print(result)
async def execute_query2():
result = await session.execute(query2)
print(result)
async def execute_subscription1():
async for result in session.subscribe(subscription1):
print(result)
async def execute_subscription2():
async for result in session.subscribe(subscription2):
print(result)
task1 = asyncio.create_task(execute_query1())
task2 = asyncio.create_task(execute_query2())
task3 = asyncio.create_task(execute_subscription1())
task4 = asyncio.create_task(execute_subscription2())
await task1
await task2
await task3
await task4Subscriptions tasks can be stopped at any time by running
task.cancel()See CONTRIBUTING.md