sgqlc.endpoint.http module

Synchronous HTTP Endpoint

This endpoint implements GraphQL client using urllib.request.urlopen() or compatible function.

This module provides command line utility:

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

Example using sgqlc.endpoint.http.HTTPEndpoint:

#!/usr/bin/env python3

import sys
import json
from sgqlc.endpoint.http import HTTPEndpoint

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 = HTTPEndpoint(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.http.HTTPEndpoint(url, base_headers=None, timeout=None, urlopen=None, method='POST')[source]

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 urllib.error.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 HTTPEndpoint._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)[source]

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.

  • 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, urlopen=None, method='POST')[source]
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).

  • urlopen – function that implements the same interface as urllib.request.urlopen(), which is used by default.

__str__()[source]

Return str(self).

_log_http_error(query, req, exc)[source]

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

Parameters:
Returns:

GraphQL-compliant dict with keys data and errors.

Return type:

dict