sgqlc.endpoint.requests module

Synchronous HTTP Endpoint using python-requests

This endpoint implements GraphQL client using the requests library.

This module provides command line utility:

$ python3 -m sgqlc.endpoint.requests http://server.com/ '{ query { ... } }'

It’s pretty much like sgqlc.endpoint.http.HTTPEndpoint, but using the requests. This comes with the convenience to use a requests.Session and requests.auth compatible authentication option (Auth tuple or callable to enable Basic/Digest/Custom HTTP Auth), which is useful when third party libraries offer such helpers (ex: requests-aws).

Example using sgqlc.endpoint.requests.RequestsEndpoint:

#!/usr/bin/env python3

import sys
import json
from sgqlc.endpoint.requests import RequestsEndpoint

try:
    token, repo = sys.argv[1:]
except ValueError:
    raise SystemExit('Usage: <token> <team/repo>')

query = '''
query GitHubRepoIssues($repoOwner: String!, $repoName: String!) {
  repository(owner: $repoOwner, name: $repoName) {
    issues(first: 100) {
      nodes {
        number
        title
      }
    }
  }
}
'''

owner, name = repo.split('/', 1)
variables = {
    'repoOwner': owner,
    'repoName': name,
}

url = 'https://api.github.com/graphql'
headers = {
    'Authorization': 'bearer ' + token,
}

endpoint = RequestsEndpoint(url, headers)
data = endpoint(query, variables)

json.dump(data, sys.stdout, sort_keys=True, indent=2, default=str)

The query may be given as bytes or str as in the example, but it may be a sgqlc.operation.Operation, which will serialize as string while also providing convenience to interepret the results.

See more examples.

license:

ISC

class sgqlc.endpoint.requests.RequestsEndpoint(url, base_headers=None, timeout=None, method='POST', auth=None, session=None)

Bases: BaseEndpoint

GraphQL access over HTTP.

This helper is very thin, just setups the correct HTTP request to GraphQL endpoint, handling logging of HTTP and GraphQL errors. The object is callable with parameters: query, variables, operation_name, extra_headers and timeout.

The user of this class should create GraphQL queries and interpret the resulting object, created from JSON data, with top level properties:

Data:

object matching the GraphQL requests, or null if only errors were returned.

Errors:

list of errors, which are objects with the key “message” and optionally others, such as “location” (for errors matching GraphQL input). Instead of raising exceptions, such as requests.exceptions.HTTPError or json.JSONDecodeError those are stored in the “exception” key.

Note

Both data and errors may be returned, for instance if a null-able field fails, it will be returned as null (Python None) in data the associated error in the array.

The class has its own logging.Logger which is used to debug, info, warning and errors. Error logging and conversion to uniform data structure similar to GraphQL, with {"errors": [...]} is done by RequestsEndpoint._log_http_error() own method, BaseEndpoint._log_json_error() and BaseEndpoint._log_graphql_error(). This last one will show the snippets of GraphQL that failed execution.

__call__(query, variables=None, operation_name=None, extra_headers=None, timeout=None)

Calls the GraphQL endpoint.

Parameters:

• query (str or bytes.) – the GraphQL query or mutation to execute. Note that this is converted using bytes(), thus one may pass an object implementing __bytes__() method to return the query, eventually in more compact form (no indentation, etc).

• variables (dict) – variables (dict) to use with query. This is only useful if the query or mutation contains $variableName. Must be a plain JSON-serializeable object (dict with string keys and values being one of dict, list, tuple, str, int, float, bool, None… – json.dumps() is used) and the keys must match exactly the variable names (no name conversion is done, no dollar-sign prefix $ should be used).

• operation_name (str) – if more than one operation is listed in query, then it should specify the one to be executed.

• extra_headers (dict) – dict with extra HTTP headers to use.

• timeout (float) – overrides the default timeout.

Returns:

dict with optional fields data containing the GraphQL returned data as nested dict and errors with an array of errors. Note that both data and errors may be returned!

Return type:

dict

__init__(url, base_headers=None, timeout=None, method='POST', auth=None, session=None)

Parameters:

• url (str) – the default GraphQL endpoint url.

• base_headers (dict) – the base HTTP headers to include in every request.

• timeout (float) – timeout in seconds to use with urllib.request.urlopen(). Optional (None uses default timeout).

• method – HTTP Method to use for the request, POST is used by default

• auth – requests.auth compatible authentication option. Auth tuple or callable to enable Basic/Digest/Custom HTTP Auth. Optional.

• session – requests.Session object. Optional.

__str__()

Return str(self).

_log_http_error(query, req, exc)

Log requests.exceptions.HTTPError, converting to GraphQL’s {"data": null, "errors": [{"message": str(exc)...}]}

Parameters:

• query (str) – the GraphQL query that triggered the result.

• req (requests.Request) – requests.Request instance that was opened.

• exc (requests.exceptions.HTTPError) – requests.exceptions.HTTPError instance

Returns:

GraphQL-compliant dict with keys data and errors.

Return type:

dict