Skip to main content

A fast and modern graphql client library designed with simplicity in mind.

Project description

Qlient: Python GraphQL Client

A fast and modern graphql client library designed with simplicity in mind.

pypi GitHub license Wheel Language

Getting Started

Installing

pip install python-qlient

Usage

This example shows a simple query

from qlient import Client

client = Client("https://countries.trevorblades.com/")

response = client.query.countries(select=["name", "capital"], where={"filter": {"code": {"eq": "CH"}}})

# which is equal to
response = client.query.countries.select(["name", "capital"]).where({"filter": {"code": {"eq": "CH"}}}).exec()

# The 'where' method is only available on queries and named like this just for synthetic sugar.
# for mutations it's 'set_variables'. 
# both methods do the same.

Documentation

The Documentation covers the following points:

Query

Queries are the A and O of graphql and are as easy as:

from qlient import Client

client = Client("https://countries.trevorblades.com/")

response = client.query.country(select=["name", "capital"], where={"filter": {"code": {"eq": "CH"}}})

Which will make a request to the server and return the name and capital of the country switzerland.

But what if you want to make a more complex query with fields within fields? Don't need to worry, we got you covered:

from qlient import Client, fields

client = Client("https://countries.trevorblades.com/")

response = client.query.country(
    select=fields("name", "capital", languages=fields("name")),  # languages is an object that needs a subselection
    where={"filter": {"code": {"eq": "CH"}}}
)

Using the fields method from qlient you can simply use *args and **kwargs for making deeper selections. By the way, you could stack this like forever.

Last but not least, what if you don't know the fields you could select? Yup, we got you somewhat covered as well. The thing is, that due to performance issues, this package is not able to completely create a query that retrieves all fields for a Query. I have set the max depth to 2 (Can be changed via Settings). This allows to still send a query without selecting any fields but you won't get them all. If you want all, use the fields function defined above.

from qlient import Client

client = Client("https://countries.trevorblades.com/")

response = client.query.country(where={"filter": {"code": {"eq": "CH"}}})

Mutation

I haven't found a real world example for making mutations without being authenticated, so here's a hypothetical one.

from qlient import Client

client = Client("https://some-host.com/authentication")

response = client.mutation.register(data={"email": "foo@bar.com", "password": "987654321"}, select=["userId"])

Subscription

Sometimes you want to execute things when something - an action - happened on the server. In those cases, you can subscribe to an event. For subscribing to an endpoint, I am using the asyncio websockets library. So have a look at their documentation for clarification.

Here is a basic example

import asyncio
from qlient import Client

client = Client("http://your-host:8080")

def on_event(data: dict):
    # ... do something with the data
    print(data)

asyncio.run(client.subscription.my_subscription(handle=on_event))  # the asyncio.run() function is important!

Different Websocket endpoint

If no websocket endpoint was specified, it gets adapted based on the given request host. for example http://localhost:3000 becomes ws://localhost:3000. Same goes for secured connections: https becomes wss. But it may be, that you have different endpoints. Therefor you can specify the websocket endpoint manually.

from qlient import Client

client = Client("http://your-host:8080", ws_endpoint="wss://your-other-host:3000")

Debugging

When you need to see the to be executed query, you simply do as following:

from qlient import Client

client = Client("https://countries.trevorblades.com/")

print(client.query.countries.query_string)

# should print something like:
# query countries { countries { code name native phone capital currency emoji emojiU continent { code name } languages { code name native rtl } states { code name } } }

And when you need to change the selection:

print(client.query.countries.select(["code", "name"]).query_string)
# which prints: query countries { countries { code name } }

Or with variables:

print(client.query.countries.select(["code", "name"]).set_variables({"filter": {"code": {"eq": "CH"}}}).query_string)
# which prints: query countries ($filter: CountryFilterInput) { countries (filter: $filter) { code name } }
# the variables are not visible in the query but rather will be send as variables dict to the server.

Transporter

For making requests, we use a transporter. (Irrelevant for Websockets.)

If none is given, a new one will be created.

Sometimes you want your own custom session to be used for making requests. For example if you need to authenticate yourself with some sort of an api key. Therefor, you can pass it directly to the transporter.

import requests

from qlient import Client, Transporter

my_session = requests.sessions.session()

my_session.headers["Authorization"] = "Bearer some-api-token"

client = Client("https://foo.bar/", transporter=Transporter(session=my_session))

AsyncTransporter

And an AsyncTransporter:

import asyncio

from qlient import Client, AsyncTransporter

client = Client("https://countries.trevorblades.com/", transporter=AsyncTransporter())

async def request_data():
    return await client.query.country(select=["name", "capital"], where={"code": "CH"})

asyncio.run(request_data())

Settings

Most things can be adjusted using the settings. When no settings are passed by to a client, the default values will be used instead

max_recursion_depth

The max_recursion_depth can be used for changing the max depth for deeper automatic selection lookup. Default is 2.

from qlient import Client, Settings

settings = Settings(max_recursion_depth=5)  # Due to performance reasons I do not recommend to go any higher than that.

client = Client("https://countries.trevorblades.com/", settings=settings)

base_response_key

The base_response_key can be changed for setting the base key that is being used to get the data from the server. Default is "data".

from qlient import Client, Settings

settings = Settings(base_response_key="my_custom_data_key")

client = Client("https://countries.trevorblades.com/", settings=settings)

base_payload_key

The base_payload_key can be changed for setting the base key that is being used to read the data from the websocket response. Default is "payload".

from qlient import Client, Settings

settings = Settings(base_payload_key="my_custom_payload_key")

client = Client("https://countries.trevorblades.com/", settings=settings)

return_requests_response

The return_requests_response can be set to True if you want the whole request back instead of just the json. Default is False.

from qlient import Client, Settings

settings = Settings(return_requests_response=True)

client = Client("https://countries.trevorblades.com/", settings=settings)

disable_selection_lookup

The disable_selection_lookup can be set to True if you want to disable the automatic selection lookup. Default is False.

from qlient import Client, Settings

settings = Settings(disable_selection_lookup=True)

client = Client("https://countries.trevorblades.com/", settings=settings)

return_full_subscription_body

The return_full_subscription_body can be set to True if you want to get the full websocket response instead of only the data.

from qlient import Client, Settings

settings = Settings(return_full_subscription_body=True)

client = Client("https://countries.trevorblades.com/", settings=settings)

Cache

The qlient packages comes with two built in caching mechanisms that are optional. Bear in mind that the cache is only used for schema introspection and queries.

MemoryCache

The memory cache uses a dictionary to store all keys and values.

from qlient import Client, MemoryCache

client = Client("https://countries.trevorblades.com/", cache=MemoryCache(ttl=5))  
# Creates a new client with a in memory cache where each object has a max time-to-live of 5 seconds.

DiskCache

The disk cache writes and reads files to and from the physical disk. Even though the disk cache is efficient, i'd recommend only using this for requests where the data doesn't change a lot.

from qlient import Client, DiskCache

client = Client("https://countries.trevorblades.com/", cache=DiskCache(ttl=60 * 60 * 24))  
# Creates a new client with a disk cache where each object has a max time-to-live of 86'400 seconds or a day.
Custom DiskCache Location

The default disk cache location is in the python-qlient folder within the system specific tempdir folder.

Windows: C:\Users\{myUser}\AppData\Local\Temp\python-qlient

Linux: /tmp/python-qlient

However, this can be changed by setting the root variable:

from qlient import DiskCache

cache = DiskCache(root=r"path/to/your/desired/root/folder")

CLI

Qlient also provides a CLI for inspecting a schema.

qlient --inspect "https://countries.trevorblades.com/"

# or short:
# qlient -i "https://countries.trevorblades.com/"

Authors

  • Daniel Seifert - Initial work - Lab9

Acknowledgments

  • Heavily inspired by Zeep

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

python-qlient-1.2.2.tar.gz (20.2 kB view hashes)

Uploaded Source

Built Distribution

python_qlient-1.2.2-py3-none-any.whl (22.8 kB view hashes)

Uploaded Python 3

Supported by

AWS AWS Cloud computing and Security Sponsor Datadog Datadog Monitoring Fastly Fastly CDN Google Google Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page