aiohttp based GraphQL client with file upload support
Project description
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 protocol for WebSocket based subscriptions
.
Requirements
- Python (>=3.7)
- aiohttp (>=3.6.0)
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
- Clone the repo
- Create and enter a dedicated virtual environment
- 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
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.