GQL

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.

WARNING: Please note that the following documentation describes the current version which is currently only available as a pre-release The documentation for the 2.x version compatible with python<3.6 is available in the 2.x branch

Installation

$ pip install gql

WARNING: Please note that the following documentation describes the current version which is currently only available as a pre-release and needs to be installed with

$ pip install --pre gql

Usage

Basic usage

from gql import gql, Client, AIOHTTPTransport

# Select your transport with a defined url endpoint
transport = AIOHTTPTransport(url="https://countries.trevorblades.com/")

# Create a GraphQL client using the defined transport
client = Client(transport=transport, fetch_schema_from_transport=True)

# Provide a GraphQL query
query = gql(
    """
    query getContinents {
      continents {
        code
        name
      }
    }
"""
)

# Execute the query on the transport
result = client.execute(query)
print(result)

Local schema validation

It is possible to validate a query locally either using a provided schema or by using introspection to get the schema from the GraphQL API server.

Using a provided schema

The schema can be provided as a String (which is usually stored in a .graphql file):

with open('path/to/schema.graphql') as f:
    schema_str = f.read()

client = Client(schema=schema_str)

OR can be created using python classes:

from .someSchema import SampleSchema
# SampleSchema is an instance of GraphQLSchema

client = Client(schema=SampleSchema)

See tests/starwars/schema.py for an example of such a schema.

Using introspection

In order to get the schema directly from the GraphQL Server API using the transport, you just need to set the fetch_schema_from_transport argument of Client to True

HTTP Headers

If you want to add additional http headers for your connection, you can specify these in your transport:

transport = AIOHTTPTransport(url='YOUR_URL', headers={'Authorization': 'token'})

GraphQL variables

You can also provide variable values with your query:

query = gql(
    """
    query getContinentName ($code: ID!) {
      continent (code: $code) {
        name
      }
    }
"""
)

params = {"code": "EU"}

# Get name of continent with code "EU"
result = client.execute(query, variable_values=params)
print(result)

params = {"code": "AF"}

# Get name of continent with code "AF"
result = client.execute(query, variable_values=params)
print(result)

GraphQL subscriptions

Using the websockets transport, it is possible to execute GraphQL subscriptions:

from gql import gql, Client, WebsocketsTransport

transport = WebsocketsTransport(url='wss://your_server/graphql')

client = Client(
    transport=transport,
    fetch_schema_from_transport=True,
)

query = gql('''
    subscription yourSubscription {
        ...
    }
''')

for result in client.subscribe(query):
    print (result)

Note: the websockets transport can also execute queries or mutations, it is not restricted to subscriptions

Execute on a local schema

It is also possible to execute queries against a local schema (so without a transport).

from gql import gql, Client

from .someSchema import SampleSchema

client = Client(schema=SampleSchema)

query = gql('''
    {
      hello
    }
''')

result = client.execute(query)

Compose GraphQL queries dynamically with the DSL module

Instead of providing the GraphQL queries as a String, it is also possible to create GraphQL queries dynamically. Using the DSL module, we can create a query using a Domain Specific Language which is created from the schema.

from gql.dsl import DSLSchema

client = Client(schema=StarWarsSchema)
ds = DSLSchema(client)

query_dsl = ds.Query.hero.select(
    ds.Character.id,
    ds.Character.name,
    ds.Character.friends.select(ds.Character.name,),
)

will create a query equivalent to:

hero {
  id
  name
  friends {
    name
  }
}

See tests/starwars/test_dsl.py for examples.

Async usage with asyncio

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

HTTP async transport

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, AIOHTTPTransport, Client
import asyncio

async def main():

    transport = AIOHTTPTransport(url='https://countries.trevorblades.com/graphql')

    # Using `async with` on the client will start a connection on the transport
    # and provide a `session` variable to execute queries on this connection
    async with Client(
        transport=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())

Websockets async transport

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, WebsocketsTransport
import asyncio

async def main():

    transport = WebsocketsTransport(url='wss://countries.trevorblades.com/graphql')

    # Using `async with` on the client will start a connection on the transport
    # and provide a `session` variable to execute queries on this connection
    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())

Websockets SSL

If you need to connect to an ssl encrypted endpoint:

• use wss instead of ws in the url of the transport

sample_transport = WebsocketsTransport(
    url='wss://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://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')

Websockets authentication

There are two ways to send authentication tokens with websockets depending on the server configuration.

1. Using HTTP Headers

sample_transport = WebsocketsTransport(
    url='wss://SERVER_URL:SERVER_PORT/graphql',
    headers={'Authorization': 'token'}
)

2. With a payload in the connection_init websocket message

sample_transport = WebsocketsTransport(
    url='wss://SERVER_URL:SERVER_PORT/graphql',
    init_payload={'Authorization': 'token'}
)

Async advanced usage

It is possible to send multiple GraphQL queries (query, mutation or subscription) in parallel, on the same websocket connection, using asyncio tasks.

In order to retry in case of connection failure, we can use the great backoff module.

# First define all your queries using a session argument:

async def execute_query1(session):
    result = await session.execute(query1)
    print(result)

async def execute_query2(session):
    result = await session.execute(query2)
    print(result)

async def execute_subscription1(session):
    async for result in session.subscribe(subscription1):
        print(result)

async def execute_subscription2(session):
    async for result in session.subscribe(subscription2):
        print(result)

# Then create a couroutine which will connect to your API and run all your queries as tasks.
# We use a `backoff` decorator to reconnect using exponential backoff in case of connection failure.

@backoff.on_exception(backoff.expo, Exception, max_time=300)
async def graphql_connection():

    transport = WebsocketsTransport(url="wss://YOUR_URL")

    client = Client(transport=transport, fetch_schema_from_transport=True)

    async with client as session:
        task1 = asyncio.create_task(execute_query1(session))
        task2 = asyncio.create_task(execute_query2(session))
        task3 = asyncio.create_task(execute_subscription1(session))
        task4 = asyncio.create_task(execute_subscription2(session))

        await asyncio.gather(task1, task2, task3, task4)

asyncio.run(graphql_connection())

Subscriptions tasks can be stopped at any time by running

task.cancel()

Contributing

See CONTRIBUTING.md

License

MIT License