Skip to content

Python aynchronous/IO GraphQL client with file upload and subscription support.

License

Notifications You must be signed in to change notification settings

DoctorJohn/aiogqlc

Repository files navigation

Asynchronous/IO GraphQL client

Versions PyPI Downloads Codecov License

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/.netlify/functions/index"

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)

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