Asyncio Support

[YassineElbouchaibi]

I am thinking of using this module and I was wondering how friendly it would be with an async application. Is there documented or undocumented support for async? If not, are there plans to add support ? Do you think just invoking the endpoint in a thread pool executor would be sufficient to avoid blocking?

The main reason I would want to use this client is code generation and typing.

Anyone ever tried sgqlc with async?

Thanks

[barbieri]

Our core does not do I/O, it's more on how to generate the operation string and how to parse the results into an easy to use structure.

We do offer some endpoints to make it simpler and none of them use asyncio yet, but it's a matter of writing that code. See https://github.com/profusion/sgqlc/blob/master/sgqlc/endpoint/http.py you may create something like that on top of the asyncio http module.

[YassineElbouchaibi]

I ended up implementing it outside the module and I import it in my own module as I am not comfortable with the typing style, documenting and testing used in this project.

To use this, one can simply add it to his/her project :

import json
import logging
from contextlib import asynccontextmanager
from typing import Dict, Optional, Union

import aiohttp
from aiohttp.client_exceptions import ClientResponseError
from sgqlc.endpoint.base import BaseEndpoint, add_query_to_url


class AIOHTTPEndpoint(BaseEndpoint):
    logger = logging.getLogger(__name__)

    def __init__(
        self,
        url: str,
        base_headers: Optional[Dict[str, str]] = None,
        timeout: Optional[int] = None,
        method: str = "POST",
        session: Optional[aiohttp.ClientSession] = None,
    ):
        self.url = url
        self.base_headers = base_headers or {}
        self.timeout = timeout
        self.method = method
        self.session = session or aiohttp.ClientSession()

    def __str__(self):
        return (
            "%s(url=%s, base_headers=%r, timeout=%r, " "method=%s, session=%s)"
        ) % (  # noqa
            self.__class__.__name__,
            self.url,
            self.base_headers,
            self.timeout,
            self.method,
            self.session.__class__,
        )

    async def __call__(
        self,
        query: Union[bytes, str],
        variables: Optional[Dict] = None,
        operation_name: Optional[str] = None,
        extra_headers: Optional[Dict[str, str]] = None,
        timeout: Optional[int] = None,
    ):
        if isinstance(query, bytes):
            query = query.decode("utf-8")
        elif not isinstance(query, str):
            # allows sgqlc.operation.Operation to be passed
            # and generate compact representation of the queries
            query = bytes(query).decode("utf-8")

        headers = self.base_headers.copy()
        if extra_headers:
            headers.update(extra_headers)

        if "Accept" not in headers:
            headers["Accept"] = "application/json; charset=utf-8"

        if self.method.upper() == "POST":
            get_http_request = self.get_http_post_request
        else:
            get_http_request = self.get_http_get_request

        req = get_http_request(
            query,
            variables,
            operation_name,
            headers,
            timeout,
        )

        self.logger.debug("Query:\n%s", query)

        try:
            async with req() as response:
                try:
                    response.raise_for_status()
                    data: Optional[Dict] = await response.json()
                    if data and data.get("errors"):
                        return self._log_graphql_error(query, data)
                    return data
                except json.JSONDecodeError as exc:
                    return self._log_json_error(await response.text(), exc)
        except aiohttp.ClientResponseError as exc:
            return self._log_http_error(exc)

    def get_http_post_request(
        self,
        query: Union[bytes, str],
        variables: Optional[Dict] = None,
        operation_name: Optional[str] = None,
        headers: Optional[Dict[str, str]] = None,
        timeout: Optional[int] = None,
    ):
        post_data = json.dumps(
            {
                "query": query,
                "variables": variables,
                "operationName": operation_name,
            }
        ).encode("utf-8")
        headers.update(
            {
                "Content-Type": "application/json; charset=utf-8",
                "Content-Length": str(len(post_data)),
            }
        )

        @asynccontextmanager
        async def post_request():
            async with self.session as session:
                async with session.post(
                    url=self.url,
                    data=post_data,
                    headers=headers,
                    timeout=timeout or self.timeout,
                ) as response:
                    yield response

        return post_request

    def get_http_get_request(
        self,
        query: Union[bytes, str],
        variables: Optional[Dict] = None,
        operation_name: Optional[str] = None,
        headers: Optional[Dict[str, str]] = None,
        timeout: Optional[int] = None,
    ):
        params = {"query": query}
        if operation_name:
            params["operationName"] = operation_name

        if variables:
            params["variables"] = json.dumps(variables)

        url = add_query_to_url(self.url, params)

        @asynccontextmanager
        async def get_request():
            async with self.session as session:
                async with session.get(
                    url=url, headers=headers, timeout=timeout or self.timeout
                ) as response:
                    yield response

        return get_request

    def _log_http_error(self, exc: ClientResponseError):
        self.logger.error("%s: %s", exc.request_info.real_url, exc)
        for h in sorted(exc.headers):
            self.logger.info("Response header: %s: %s", h, exc.headers[h])

        return {
            "data": None,
            "errors": [
                {
                    "message": str(exc),
                    "exception": exc,
                    "status": exc.code,
                    "headers": exc.headers,
                }
            ],
        }


if __name__ == "__main__":  # pragma: no cover
    import asyncio

    async def main():
        import argparse
        import sys

        def tuple_arg(s):
            p = s.split("=", 1)
            if len(p) == 2:
                return p
            return s.split(":", 1)

        ap = argparse.ArgumentParser(
            description="Access GraphQL endpoint using HTTP",
        )

        # Generic options to access the GraphQL API
        ap.add_argument("url", help="GraphQL endpoint address.")
        ap.add_argument("query", help="GraphQL query or mutation.")

        ap.add_argument(
            "--var",
            "-V",
            action="append",
            type=tuple_arg,
            help="NAME=VALUE variable to send alongside query.",
            default=[],
        )
        ap.add_argument(
            "--operation",
            "-o",
            help="GraphQL operation name to send alongside query.",
            default=None,
        )
        ap.add_argument(
            "--header",
            "-H",
            action="append",
            type=tuple_arg,
            help=(
                "NAME=VALUE HTTP header to send. "
                'Example: "Authorization: bearer ${token}"'
            ),
            default=[],
        )
        ap.add_argument(
            "--timeout",
            "-t",
            type=float,
            default=None,
            help="Request timeout, in seconds.",
        )
        ap.add_argument(
            "--verbose",
            "-v",
            help="Increase verbosity",
            action="count",
            default=0,
        )
        ap.add_argument(
            "--method",
            "-m",
            choices=["GET", "POST"],
            help="HTTP Method to use",
            default="POST",
        )

        args = ap.parse_args()

        logfmt = "%(levelname)s: %(message)s"
        logging.basicConfig(
            format=logfmt,
            level=max(10, 40 - (args.verbose * 10)),
        )

        endpoint = AIOHTTPEndpoint(
            args.url, dict(args.header), args.timeout, method=args.method
        )
        data = await endpoint(args.query, dict(args.var))

        json.dump(data, sys.stdout, sort_keys=True, indent=2, default=str)
        sys.stdout.write("\n")
        if data.get("errors"):
            sys.exit(1)

    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())


__all__ = ["AIOHTTPEndpoint"]

then import AIOHTTPEndpoint from it and use it:

import asyncio

async def main():
    from generated.graphql.list_countries_and_languages import Operations
    from python_sandbox.utils.sgqlc import AIOHTTPEndpoint

    op = Operations.query.list_countries_and_languages

    print(op)

    url = "https://countries.trevorblades.com/"
    endpoint = AIOHTTPEndpoint(url, method="GET")
    data = await endpoint(op)
    data = op + data

    print(data.countries)

loop = asyncio.get_event_loop()
loop.run_until_complete(main())

One can also use it as a cli:

python sgqlc/aiohttp.py "https://countries.trevorblades.com/" "query ListCountriesAndLanguages {
  countries {
    name
  }
  languages {
    name
  }
}"

Edit: Pretty much all of it was tested except _log_http_error, For those that want to use this code. do it at your own risk.

[barbieri]

thank you so much! I didn't test, but a quick look it looks correct and I'd love to see this as a PR to the project.

I know coding style and some preferences may diverge, but we have flake8 and some checkers. It's alright to add the typing notations... we lack in the rest of the project because I started it when typing was not widely used in Python community... but I do not oppose adding it (we're using it for new projects for our customers).

[marcovc]

Hi Yassine and barbieri, and thanks for your code.

Doesn't

async with self.session as session:

inside "post_request" means that the session will be closed after the first call to post ?

[barbieri]

need to check, I'm not sure but could be the case... but maybe it's ref-counted and won't close if there are other users?

[geekscrapy]

Was this ever included? Really interested in using it as part of the core lib