Source code for kirjava.client

"""Contains the Client class itself."""

import json
import requests
import time
from .utilities import files_to_map, get_files_from_variables, create_response_error_message, pack_files

[docs]class Client: """A GraphQL client. This is the object which sends requests to the GraphQL server. The URL that serves the GraphQL content is given on creating the Client. :param str url: The URL of the GraphQL server to interact with. :param dict headers: Any additional HTTP headers.""" def __init__(self, url): self._url = url self._headers = { "Accept": "application/json", "Content-Type": "application/json" } self._history = [] self.session = requests.Session() def __repr__(self): return f"<Client (URL: {self._url})>" @property def url(self): """The URL of the GraphQL server to interact with. :rtype: ``str``""" return self._url @property def headers(self): """The HTTP headers that will be sent with every request. :rtype: ``dict``""" return self._headers @property def history(self): """The queries sent, most recent first. :rtype: ``tuple``""" return tuple(self._history)
[docs] def execute(self, message, method="POST", variables=None, retries=0, retry_statuses=None): """Sends a request to the GraphQL server. :param str message: The query to make. :param str method: By default, POST requests are sent, but this can be\ overriden here. :param dict variables: Any GraphQL variables can be passed here. :param int retries: The number of times to retry on failure. :param list retry_statuses: The HTTP statuses to retry on. :rtype: ``dict``""" headers = {key: value for key, value in self._headers.items()} variables, files = get_files_from_variables(variables) operation = json.dumps({"variables": variables, "query": message}) if files: del headers["Content-Type"] data = {"operations": operation, "map": json.dumps(files_to_map(files))} files = pack_files(files) response = self.request_with_retries( operation=data, headers=headers, method=method, retries=retries, retry_statuses=retry_statuses, files=files ) else: response = self.request_with_retries( operation=operation, headers=headers, method=method, retries=retries, retry_statuses=retry_statuses ) try: result = response.json() except json.decoder.JSONDecodeError: raise ValueError(create_response_error_message(response)) self._history.insert(0, ( {"query": message, "variables": variables or {}}, result )) return result
[docs] def request_with_retries(self, operation, headers, files=None, method="POST", retries=0, retry_statuses=None): """Sends a GraphQL request, retrying if necessary the specified number of times. :param str operation: The GraphQL operation to send. :param dict headers: The HTTP headers to send. :param dict files: The files to send. :param str method: The HTTP method to use. :param int retries: The number of times to retry. :param list retry_statuses: The HTTP statuses to retry on. :rtype: ``requests.Response``""" attempts = 0 while True: try: response = self.session.request( method, self._url, headers=headers, data=operation, files=files ) if retry_statuses and response.status_code in retry_statuses: raise Exception(f"Status code {response.status_code}") return response except Exception as e: attempts += 1 if attempts > retries: raise e time.sleep(2**attempts) continue