pygqlc 3.2.0

Python client for graphql APIs

pygqlc

Python client for graphql APIs

Scope

This is an open source project, please feel free to fork it and PR to contribute with the community! Repo for the project: https://github.com/valiot/pygqlc

Installation

Requirements:

• Python 3.9+
• Pipenv

Install directly from pypi:

$ cd <my-project-dir>
$ pipenv --python 3.7 # or 3.6
$ pipenv shell
$ pipenv install pygqlc
$ python
$ >> import pygqlc
$ >> print(pygqlc.name)

If you get "pygqlc" printed in the python repl, the installation succeded!

Usage

import os
from pygqlc import GraphQLClient
gql = GraphQLClient()
gql.addEnvironment(
    'dev',
    url=os.environ.get('API'), # should be an https url
    wss=os.environ.get('WSS'), # should be an ws/wss url
    headers={'Authorization': os.environ.get('TOKEN')},
    default=True)

From now on, you can access to the main API:

gql.query, gql.mutate, gql.subscribe

For queries:

query = '''
query{
  authors{
    name
  }
}
'''
data, errors = gql.query( query )

For mutations:

create_author = '''
mutation {
  createAuthor(){
    successful
    messages{field message}
    result{id insertedAt}
  }
}
'''
data, errors = gql.mutate( create_author )

For subscriptions:

def on_auth_created(message):
  print(message)

sub_author_created = '''
subscription{
  authorCreated{
    successful
    messages{field message}
    result{id insertedAt}
  }
}
'''
# unsub may be None if subscription fails (no internet connection, host unreachable, bad subscription doc, etc)
unsub = gql.subscribe(sub_author_created, callback=on_auth_created)
...
# when finishing the subscription:
unsub()
# when finishing all subscriptions:
gql.close()

The subscribe method, returns an unsubscribe function, this allows to stop subscriptions whenever needed.

After finishing all subscriptions, the method GraphQLClient.close() should be called to close correctly the open GQL/websocket connections.

To reset all subscriptions and websocket connection use the method GraphQLClient.resetSubsConnection().

To be noted:

All main methods from the API accept a variables param. it is a dictionary type and may include variables from your queries or mutations:

query_with_vars = '''
query CommentsFromAuthor(
  $authorName: String!
  $limit: Int
){
  author(
    findBy:{ name: $authorName }
  ){
    id
    name
    comments(
      orderBy:{desc: ID}
      limit: $limit
    ){
      id
      blogPost{name}
      body
    }
  }
}
'''

data, errors = gql.query(
  query=query_with_vars,
  variables={
    "authorName": "Baruc",
    "limit": 10
  }
)

There is also an optional parameter flatten that simplifies the response format:

# From this:
response = {
  'data': {
    'authors': [
      { 'name': 'Baruc' },
      { 'name': 'Juan' },
      { 'name': 'Gerardo' }
    ]
  }
}
# To this:
authors = [
  { 'name': 'Baruc' },
  { 'name': 'Juan' },
  { 'name': 'Gerardo' }
]

Simplifying the data access from this:

response['data']['authors'][0]['name']

to this:

authors[0]['name']

It is query(query, variables, flatten=True) by default, to avoid writing it down everytime

The (_, errors) part of the response, is the combination of GraphQL errors, and communication errors, simplifying validations, it has this form:

errors = [
  {"field": <value>, "message":<msg>},
  {"field": <value>, "message":<msg>},
  {"field": <value>, "message":<msg>},
  ...
]

The field Attribute it's only available for GraphQL errors, when it is included in the response, so it's suggested that every mutation has at least this parameters in the response:

mutation{
  myMutation(<mutationParams>){
    successful
    messages{
      field
      message
    }
    result{
      <data of interest>
    }
  }
}

Post timeout:

You can set a post timeout to avoid an inactive process.

Use gql.setPostTimeout(seconds), or directly in the environment gql.addEnvironment(post_timeout=seconds). Default port_timeout is 60 seconds

Websocket timeout:

You can set a websocket timeout to keep subscriptions alive.

Use gql.setTimeoutWebsocket(seconds), or directly in the environment gql.addEnvironment(timeoutWebsocket=seconds). Default timeoutWebsocket is 60 seconds

for mantainers:

Initial configuration

first of all, ensure you have configured poetry repositories correctly: poetry config repositories.valiot https://pypi.valiot.io/

and their credentials:

For private valiot's pypi:

poetry config http-basic.valiot <username> <password>

(ask adrian to send you the proper username and password for this step)

And for public pypi:

poetry config pypi-token.pypi <pypi-token>

(ask adrian or baruc to generate a token for you)

then,

regular publish steps (after initial configuration)

deploy using:

poetry version <patch | minor | major>

then publish to valiot's private pypi:

poetry publish --build -r valiot # build and PUBLISH TO PRIVATE VALIOTs PYPI

or:

poetry publish -r valiot

(if you already built the package)

then publish to public pypi:

poetry publish

After release, publish to github:

cat pygqlc/__version__.py

gh release create

gh release upload v<#.#.#> ./dist/pygqlc-<#.#.#>-py3-none-any.whl

and don't forget to keep the CHANGELOG.md updated!