Asynchronous/IO GraphQL client

A Python asynchronous/IO GraphQL client based on aiohttp.

In addition to standard HTTP POST queries and mutations this client fully supports the GraphQL multipart form requests spec for file uploads and the graphql-ws subprotocol for WebSocket based subscriptions.

Read the documentation

Installation

pip install aiogqlc

Basic usage

Check the documentation for detailed and more advanced usage examples.

Queries

import asyncio
import aiohttp
from aiogqlc import GraphQLClient

endpoint = "https://swapi-graphql.netlify.app/graphql"

document = """
    query {
        allFilms {
            films {
                title
            }
        }
    }
"""


async def main():
    async with aiohttp.ClientSession() as session:
        client = GraphQLClient(endpoint, session=session)
        response = await client.execute(document)
        print(await response.json())


if __name__ == "__main__":
    asyncio.run(main())

Mutations

import aiohttp
from aiogqlc import GraphQLClient

document = """
    mutation ($userId: ID!) {
        deleteUser (id: $userId) {
            id
        }
    }
"""

variables = {
    "userId": "42",
}


async def main():
    async with aiohttp.ClientSession() as session:
        client = GraphQLClient("https://example.com/graphql/", session=session)
        response = await client.execute(document, variables=variables)
        print(await response.json())

File uploads

import aiohttp
from aiogqlc import GraphQLClient

document = """
    mutation($file: Upload!) {
        uploadFile(file: $file) {
            size
        }
    }
"""

variables = {
    "file": open("test.txt", "rb")
}


async def foo():
    async with aiohttp.ClientSession() as session:
        client = GraphQLClient("https://example.com/graphql/", session=session)
        response = await client.execute(document, variables=variables)
        print(await response.json())

Subscriptions

import aiohttp
from aiogqlc import GraphQLClient

document = """
    subscription($postId: ID!) {
        likeAdded(postId: $postId)
    }
"""

variables = {
    "postId": "42"
}


async def main():
    async with aiohttp.ClientSession() as session:
        client = GraphQLClient("https://example.com/graphql/", session=session)

        async with client.connect() as connection:
            async for payload in connection.subscribe(document, variables=variables):
                print(payload)

Documentation

Read the documentation to learn more about queries, mutations, subscriptions, file uploads and even authentication.