Skip to main content

aiohttp based GraphQL client with file upload support

Project description

Asynchronous/IO GraphQL client

PyPI Downloads License Tests

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 protocol for WebSocket based subscriptions.

Requirements

Installation

pip install aiogqlc

Usage

Executing simple queries

import asyncio
import aiohttp
from aiogqlc import GraphQLClient

ENDPOINT = "https://swapi-graphql.netlify.app/.netlify/functions/index"

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

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


if __name__ == "__main__":
    asyncio.get_event_loop().run_until_complete(main())

Adding authorization headers

import aiohttp
from aiogqlc import GraphQLClient

headers = {
    'Authorization': 'Token <your-token-here>'
}

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

Single file upload

import aiohttp
from aiogqlc import GraphQLClient

query = '''
    mutation($file: Upload!) {
        yourSingleUploadMutation(file: $file) {
            errors {
                field
                messages
            }
        }
    }
'''

variables = {
    'file': open('test.zip', 'rb'),
}

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

File list upload

import aiohttp
from aiogqlc import GraphQLClient

query = '''
    mutation($files: [Upload!]!) {
        yourMultiUploadMutation(files: $files) {
            errors {
                field
                messages
            }
        }
    }
'''

variables = {
    'files': [
        open('foo.zip', 'rb'),
        open('var.zip', 'rb'),
    ],
}

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

Selecting an operation

import aiohttp
from aiogqlc import GraphQLClient

query = '''
    query Operation1 {
        allFilms {
            id
        }
    }
    query Operation2 {
        film(id: 1) {
            id
        }
    }
'''

async def foo():
    async with aiohttp.ClientSession() as session:
        client = GraphQLClient('https://example.com/graphql/', session=session)
        response = await client.execute(query, operation='Operation2')
        print(await response.json())

Starting a subscription

import aiohttp
from aiogqlc import GraphQLClient


query = """
    subscription CommentAdded($article: ID!) {
        commentAdded(article: $article) {
            id
            content
        }
    }
"""

variables = {
    "article": "42"
}


async def foo():
    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(query, variables=variables):
                print(payload)

Start multiple subscriptions using a single connection

The graphql-ws protocol allows us to reuse a single WebSocket connection for multiple subscriptions. While the example below shows running two subscriptions sequential, running multiple subscriptions over one connection in parallel works as well.

import aiohttp
from aiogqlc import GraphQLClient


query = """
    subscription ItemAdded($list: ID!) {
        itemAdded(list: $list) {
            id
            content
        }
    }
"""


async def foo():
    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(query, variables={"list": "1"}):
                print(payload)

            async for payload in connection.subscribe(query, variables={"list": "2"}):
                print(payload)

Selecting a subscription operating

import aiohttp
from aiogqlc import GraphQLClient


query = """
    subscription Subscription1 {
        count(to: 11)
    }
    subscription Subscription2 {
        count(to: 22)
    }
"""


async def foo():
    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(query, operation="Subscription2"):
                print(payload)

Using connection params

Some servers supporting graphql-ws allow clients to specify connection params.

import aiohttp
from aiogqlc import GraphQLClient


query = """
    subscription {
        userCreated
    }
"""


async def foo():
    async with aiohttp.ClientSession() as session:
        client = GraphQLClient('https://example.com/graphql/', session=session)
        
        connection_params = {
            "username": "john",
            "password": "1234",
            "keep_alive_interval": 20,
        }

        async with client.connect(params=connection_params) as connection:
            async for payload in connection.subscribe(query):
                print(payload)

Contributing

Quickstart

  1. Clone the repo
  2. Create and enter a dedicated virtual environment
  3. Run make dev to install dev dependencies and a pre-commit hook that automatically formats code.

If you do not have make installed, either install it or take a look at the Makefile to find out how to run individual commands manually.

Formatting and linting

  • Run make format to format the code.
  • Run make lint to lint the code.

Testing

  • Run make test to run all tests in your local environment.
  • Run make tox to run all tests with all supported python versions.

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

aiogqlc-3.0.0.tar.gz (6.5 kB view hashes)

Uploaded source

Built Distribution

aiogqlc-3.0.0-py3-none-any.whl (7.4 kB view hashes)

Uploaded py3

Supported by

AWS AWS Cloud computing Datadog Datadog Monitoring Facebook / Instagram Facebook / Instagram PSF Sponsor Fastly Fastly CDN Google Google Object Storage and Download Analytics Huawei Huawei PSF Sponsor Microsoft Microsoft PSF Sponsor NVIDIA NVIDIA PSF Sponsor Pingdom Pingdom Monitoring Salesforce Salesforce PSF Sponsor Sentry Sentry Error logging StatusPage StatusPage Status page