Source code for gqlmod.helpers.aiohttp

Helpers for using :py:mod:`aiohttp` to build a provider.

Requires the ``aiohttp`` extra.
import aiohttp
import graphql

[docs]class AiohttpProvider: """ Help build an HTTP-based provider based on aiohttp. You should fill in :py:attr:`endpoint` and possibly override :py:meth:`modify_request_args()`. """ #: The URL to send requests to. endpoint: str #: Timeout policy to use, if any. timeout: aiohttp.ClientTimeout = None #: Whether a JSON-based or form-like request should be used. use_json: bool = False
[docs] def modify_request_args(self, variables, kwargs): """ Apply policies about the request, primarily authentication. """
_session = None @property def session(self): if self._session is None: self._session = aiohttp.ClientSession() # Only needs to be a context manager for cleanup reasons. return self._session async def __call__(self, query, variables): payload = { 'query': query, 'variables': variables or {} } data_key = 'json' if self.use_json else 'data' kwargs = { 'timeout': self.timeout, data_key: payload, } self.modify_request_args(variables, kwargs) resp = await, **kwargs) result = await resp.json() resp.raise_for_status() assert 'errors' in result or 'data' in result, 'Received non-compatible response "{}"'.format(result) return graphql.ExecutionResult( errors=result.get('errors'), data=result.get('data') )