def gql(request_string: str) -> DocumentNode:
    """Given a String containing a GraphQL request, parse it into a Document.

    :param request_string: the GraphQL request as a String
    :type request_string: str
    :return: a Document which can be later executed or subscribed by a
        :class:`Client <gql.client.Client>`, by an
        :class:`async session <gql.client.AsyncClientSession>` or by a
        :class:`sync session <gql.client.SyncClientSession>`

    :raises GraphQLError: if a syntax error is encountered.
    """
    source = Source(request_string, "GraphQL request")
    return parse(source) class Client:
    """The Client class is the main entrypoint to execute GraphQL requests
    on a GQL transport.

    It can take sync or async transports as argument and can either execute
    and subscribe to requests itself with the
    :func:`execute <gql.client.Client.execute>` and
    :func:`subscribe <gql.client.Client.subscribe>` methods
    OR can be used to get a sync or async session depending on the
    transport type.

    To connect to an :ref:`async transport <async_transports>` and get an
    :class:`async session <gql.client.AsyncClientSession>`,
    use :code:`async with client as session:`

    To connect to a :ref:`sync transport <sync_transports>` and get a
    :class:`sync session <gql.client.SyncClientSession>`,
    use :code:`with client as session:`
    """

    def __init__(
        self,
        schema: Optional[Union[str, GraphQLSchema]] = None,
        introspection=None,
        transport: Optional[Union[Transport, AsyncTransport]] = None,
        fetch_schema_from_transport: bool = False,
        execute_timeout: Optional[Union[int, float]] = 10,
        serialize_variables: bool = False,
        parse_results: bool = False,
    ):
        """Initialize the client with the given parameters.

        :param schema: an optional GraphQL Schema for local validation
                See :ref:`schema_validation`
        :param transport: The provided :ref:`transport <Transports>`.
        :param fetch_schema_from_transport: Boolean to indicate that if we want to fetch
                the schema from the transport using an introspection query
        :param execute_timeout: The maximum time in seconds for the execution of a
                request before a TimeoutError is raised. Only used for async transports.
                Passing None results in waiting forever for a response.
        :param serialize_variables: whether the variable values should be
            serialized. Used for custom scalars and/or enums. Default: False.
        :param parse_results: Whether gql will try to parse the serialized output
                sent by the backend. Can be used to unserialize custom scalars or enums.
        """

        if introspection:
            assert (
                not schema
            ), "Cannot provide introspection and schema at the same time."
            schema = build_client_schema(introspection)

        if isinstance(schema, str):
            type_def_ast = parse(schema)
            schema = build_ast_schema(type_def_ast)

        if transport and fetch_schema_from_transport:
            assert (
                not schema
            ), "Cannot fetch the schema from transport if is already provided."

            assert not type(transport).__name__ == "AppSyncWebsocketsTransport", (
                "fetch_schema_from_transport=True is not allowed "
                "for AppSyncWebsocketsTransport "
                "because only subscriptions are allowed on the realtime endpoint."
            )

        if schema and not transport:
            transport = LocalSchemaTransport(schema)

        # GraphQL schema
        self.schema: Optional[GraphQLSchema] = schema

        # Answer of the introspection query
        self.introspection = introspection

        # GraphQL transport chosen
        self.transport: Optional[Union[Transport, AsyncTransport]] = transport

        # Flag to indicate that we need to fetch the schema from the transport
        # On async transports, we fetch the schema before executing the first query
        self.fetch_schema_from_transport: bool = fetch_schema_from_transport

        # Enforced timeout of the execute function (only for async transports)
        self.execute_timeout = execute_timeout

        self.serialize_variables = serialize_variables
        self.parse_results = parse_results

    def validate(self, document: DocumentNode):
        """:meta private:"""
        assert (
            self.schema
        ), "Cannot validate the document locally, you need to pass a schema."

        validation_errors = validate(self.schema, document)
        if validation_errors:
            raise validation_errors[0]

    def execute_sync(self, document: DocumentNode, *args, **kwargs) -> Dict:
        """:meta private:"""
        with self as session:
            return session.execute(document, *args, **kwargs)

    async def execute_async(self, document: DocumentNode, *args, **kwargs) -> Dict:
        """:meta private:"""
        async with self as session:
            return await session.execute(document, *args, **kwargs)

    def execute(self, document: DocumentNode, *args, **kwargs) -> Dict:
        """Execute the provided document AST against the remote server using
        the transport provided during init.

        This function **WILL BLOCK** until the result is received from the server.

        Either the transport is sync and we execute the query synchronously directly
        OR the transport is async and we execute the query in the asyncio loop
        (blocking here until answer).

        This method will:

         - connect using the transport to get a session
         - execute the GraphQL request on the transport session
         - close the session and close the connection to the server

         If you have multiple requests to send, it is better to get your own session
         and execute the requests in your session.

         The extra arguments passed in the method will be passed to the transport
         execute method.
        """

        if isinstance(self.transport, AsyncTransport):

            # Get the current asyncio event loop
            # Or create a new event loop if there isn't one (in a new Thread)
            try:
                with warnings.catch_warnings():
                    warnings.filterwarnings(
                        "ignore", message="There is no current event loop"
                    )
                    loop = asyncio.get_event_loop()
            except RuntimeError:
                loop = asyncio.new_event_loop()
                asyncio.set_event_loop(loop)

            assert not loop.is_running(), (
                "Cannot run client.execute(query) if an asyncio loop is running."
                " Use 'await client.execute_async(query)' instead."
            )

            data: Dict[Any, Any] = loop.run_until_complete(
                self.execute_async(document, *args, **kwargs)
            )

            return data

        else:  # Sync transports
            return self.execute_sync(document, *args, **kwargs)

    async def subscribe_async(
        self, document: DocumentNode, *args, **kwargs
    ) -> AsyncGenerator[Dict, None]:
        """:meta private:"""
        async with self as session:

            generator: AsyncGenerator[Dict, None] = session.subscribe(
                document, *args, **kwargs
            )

            async for result in generator:
                yield result

    def subscribe(
        self, document: DocumentNode, *args, **kwargs
    ) -> Generator[Dict, None, None]:
        """Execute a GraphQL subscription with a python generator.

        We need an async transport for this functionality.
        """

        # Get the current asyncio event loop
        # Or create a new event loop if there isn't one (in a new Thread)
        try:
            with warnings.catch_warnings():
                warnings.filterwarnings(
                    "ignore", message="There is no current event loop"
                )
                loop = asyncio.get_event_loop()
        except RuntimeError:
            loop = asyncio.new_event_loop()
            asyncio.set_event_loop(loop)

        async_generator = self.subscribe_async(document, *args, **kwargs)

        assert not loop.is_running(), (
            "Cannot run client.subscribe(query) if an asyncio loop is running."
            " Use 'await client.subscribe_async(query)' instead."
        )

        try:
            while True:
                # Note: we need to create a task here in order to be able to close
                # the async generator properly on python 3.8
                # See https://bugs.python.org/issue38559
                generator_task = asyncio.ensure_future(
                    async_generator.__anext__(), loop=loop
                )
                result = loop.run_until_complete(generator_task)
                yield result

        except StopAsyncIteration:
            pass

        except (KeyboardInterrupt, Exception):

            # Graceful shutdown by cancelling the task and waiting clean shutdown
            generator_task.cancel()

            try:
                loop.run_until_complete(generator_task)
            except (StopAsyncIteration, asyncio.CancelledError):
                pass

            # Then reraise the exception
            raise

    async def __aenter__(self):

        assert isinstance(
            self.transport, AsyncTransport
        ), "Only a transport of type AsyncTransport can be used asynchronously"

        await self.transport.connect()

        if not hasattr(self, "session"):
            self.session = AsyncClientSession(client=self)

        # Get schema from transport if needed
        try:
            if self.fetch_schema_from_transport and not self.schema:
                await self.session.fetch_schema()
        except Exception:
            # we don't know what type of exception is thrown here because it
            # depends on the underlying transport; we just make sure that the
            # transport is closed and re-raise the exception
            await self.transport.close()
            raise

        return self.session

    async def __aexit__(self, exc_type, exc, tb):

        await self.transport.close()

    def __enter__(self):

        assert not isinstance(self.transport, AsyncTransport), (
            "Only a sync transport can be used."
            " Use 'async with Client(...) as session:' instead"
        )

        self.transport.connect()

        if not hasattr(self, "session"):
            self.session = SyncClientSession(client=self)

        # Get schema from transport if needed
        try:
            if self.fetch_schema_from_transport and not self.schema:
                self.session.fetch_schema()
        except Exception:
            # we don't know what type of exception is thrown here because it
            # depends on the underlying transport; we just make sure that the
            # transport is closed and re-raise the exception
            self.transport.close()
            raise

        return self.session

    def __exit__(self, *args):
        self.transport.close() def get_parser(with_examples: bool = False) -> ArgumentParser:
    """Provides an ArgumentParser for the gql-cli script.

    This function is also used by sphinx to generate the script documentation.

    :param with_examples: set to False by default so that the examples are not
                          present in the sphinx docs (they are put there with
                          a different layout)
    """

    parser = ArgumentParser(
        description=description,
        epilog=examples if with_examples else None,
        formatter_class=RawDescriptionHelpFormatter,
    )
    parser.add_argument(
        "server", help="the server url starting with http://, https://, ws:// or wss://"
    )
    parser.add_argument(
        "-V",
        "--variables",
        nargs="*",
        help="query variables in the form key:json_value",
    )
    parser.add_argument(
        "-H", "--headers", nargs="*", help="http headers in the form key:value"
    )
    parser.add_argument("--version", action="version", version=f"v{__version__}")
    group = parser.add_mutually_exclusive_group()
    group.add_argument(
        "-d",
        "--debug",
        help="print lots of debugging statements (loglevel==DEBUG)",
        action="store_const",
        dest="loglevel",
        const=logging.DEBUG,
    )
    group.add_argument(
        "-v",
        "--verbose",
        help="show low level messages (loglevel==INFO)",
        action="store_const",
        dest="loglevel",
        const=logging.INFO,
    )
    parser.add_argument(
        "-o",
        "--operation-name",
        help="set the operation_name value",
        dest="operation_name",
    )
    parser.add_argument(
        "--print-schema",
        help="get the schema from instrospection and print it",
        action="store_true",
        dest="print_schema",
    )
    parser.add_argument(
        "--transport",
        default="auto",
        choices=[
            "auto",
            "aiohttp",
            "phoenix",
            "websockets",
            "appsync_http",
            "appsync_websockets",
        ],
        help=(
            "select the transport. 'auto' by default: "
            "aiohttp or websockets depending on url scheme"
        ),
        dest="transport",
    )

    appsync_description = """
By default, for an AppSync backend, the IAM authentication is chosen.

If you want API key or JWT authentication, you can provide one of the
following arguments:"""

    appsync_group = parser.add_argument_group(
        "AWS AppSync options", description=appsync_description
    )

    appsync_auth_group = appsync_group.add_mutually_exclusive_group()

    appsync_auth_group.add_argument(
        "--api-key", help="Provide an API key for authentication", dest="api_key",
    )

    appsync_auth_group.add_argument(
        "--jwt", help="Provide an JSON Web token for authentication", dest="jwt",
    )

    return parser async def main(args: Namespace) -> int:
    """Main entrypoint of the gql-cli script

    :param args: The parsed command line arguments
    :return: The script exit code (0 = ok, 1 = error)
    """

    # Set requested log level
    if args.loglevel is not None:
        logging.basicConfig(level=args.loglevel)

    try:
        # Instantiate transport from command line arguments
        transport = get_transport(args)

        if transport is None:
            return 1

        # Get extra execute parameters from command line arguments
        # (variables, operation_name)
        execute_args = get_execute_args(args)

    except ValueError as e:
        print(f"Error: {e}", file=sys.stderr)
        return 1

    # By default, the exit_code is 0 (everything is ok)
    exit_code = 0

    # Connect to the backend and provide a session
    async with Client(
        transport=transport, fetch_schema_from_transport=args.print_schema
    ) as session:

        if args.print_schema:
            schema_str = print_schema(session.client.schema)
            print(schema_str)

            return exit_code

        while True:

            # Read multiple lines from input and trim whitespaces
            # Will read until EOF character is received (Ctrl-D)
            query_str = sys.stdin.read().strip()

            # Exit if query is empty
            if len(query_str) == 0:
                break

            # Parse query, continue on error
            try:
                query = gql(query_str)
            except GraphQLError as e:
                print(e, file=sys.stderr)
                exit_code = 1
                continue

            # Execute or Subscribe the query depending on transport
            try:
                try:
                    async for result in session.subscribe(query, **execute_args):
                        print(json.dumps(result))
                except KeyboardInterrupt:  # pragma: no cover
                    pass
                except NotImplementedError:
                    result = await session.execute(query, **execute_args)
                    print(json.dumps(result))
            except (GraphQLError, TransportQueryError) as e:
                print(e, file=sys.stderr)
                exit_code = 1

    return exit_code class TransportAlreadyConnected(TransportError):
    """Transport is already connected.

    Exception generated when the client is trying to connect to the transport
    while the transport is already connected.
    """ class TransportClosed(TransportError):
    """Transport is already closed.

    This exception is generated when the client is trying to use the transport
    while the transport was previously closed.
    """ class TransportProtocolError(TransportError):
    """Transport protocol error.

    The answer received from the server does not correspond to the transport protocol.
    """ class TransportQueryError(Exception):
    """The server returned an error for a specific query.

    This exception should not close the transport connection.
    """

    query_id: Optional[int]
    errors: Optional[List[Any]]
    data: Optional[Any]
    extensions: Optional[Any]

    def __init__(
        self,
        msg: str,
        query_id: Optional[int] = None,
        errors: Optional[List[Any]] = None,
        data: Optional[Any] = None,
        extensions: Optional[Any] = None,
    ):
        super().__init__(msg)
        self.query_id = query_id
        self.errors = errors
        self.data = data
        self.extensions = extensions class TransportServerError(TransportError):
    """The server returned a global error.

    This exception will close the transport connection.
    """

    code: Optional[int]

    def __init__(self, message: str, code: Optional[int] = None):
        super(TransportServerError, self).__init__(message)
        self.code = code class TemporaryFile:
    """Class used to generate temporary files for the tests"""

    def __init__(self, content: Union[str, bytearray]):

        mode = "w" if isinstance(content, str) else "wb"

        # We need to set the newline to '' so that the line returns
        # are not replaced by '\r\n' on windows
        newline = "" if isinstance(content, str) else None

        self.file = tempfile.NamedTemporaryFile(
            mode=mode, newline=newline, delete=False
        )

        with self.file as f:
            f.write(content)

    @property
    def filename(self):
        return self.file.name

    def __enter__(self):
        return self

    def __exit__(self, type, value, traceback):
        os.unlink(self.filename)
import io
import json
import pytest
    import aiofiles
from gql import Client, gql
from gql.cli import get_parser, main
from gql.transport.exceptions import (
    TransportAlreadyConnected,
    TransportClosed,
    TransportProtocolError,
    TransportQueryError,
    TransportServerError,
)
from .conftest import TemporaryFile
    from aiohttp import web
    from gql.transport.aiohttp import AIOHTTPTransport
    from aiohttp import web
    from gql.transport.aiohttp import AIOHTTPTransport
    from aiohttp import web
    from gql.transport.aiohttp import AIOHTTPTransport
    from aiohttp import web
    from gql.transport.aiohttp import AIOHTTPTransport
    from aiohttp import web
    from gql.transport.aiohttp import AIOHTTPTransport
    from aiohttp import web
    from gql.transport.aiohttp import AIOHTTPTransport
    from aiohttp import web
    from gql.transport.aiohttp import AIOHTTPTransport
    from aiohttp import web
    from gql.transport.aiohttp import AIOHTTPTransport
    from aiohttp import web
    from gql.transport.aiohttp import AIOHTTPTransport
    from aiohttp import web
    from gql.transport.aiohttp import AIOHTTPTransport
    from aiohttp import web
    from gql.transport.aiohttp import AIOHTTPTransport
    from aiohttp import DummyCookieJar
    from aiohttp import web
    from gql.transport.aiohttp import AIOHTTPTransport
    from aiohttp import web
    from gql.transport.aiohttp import AIOHTTPTransport
    from aiohttp import web
    from gql.transport.aiohttp import AIOHTTPTransport
    from aiohttp import web
    from aiohttp import web
    from gql.transport.aiohttp import AIOHTTPTransport
    from aiohttp import web
    from gql.transport.aiohttp import AIOHTTPTransport
    from aiohttp import web
    from aiohttp import web
    from gql.transport.aiohttp import AIOHTTPTransport
    from aiohttp import web, ClientSession
    from gql.transport.aiohttp import AIOHTTPTransport
    from aiohttp import web
    from gql.transport.aiohttp import AIOHTTPTransport
    from aiohttp import web
    from gql.transport.aiohttp import AIOHTTPTransport
    from aiohttp import web
    from gql.transport.aiohttp import AIOHTTPTransport
    from aiohttp import web
    from aiohttp import web
    from aiohttp import web
    from aiohttp import web
    from gql.transport.aiohttp import AIOHTTPTransport
    from aiohttp import web
    from gql.transport.aiohttp import AIOHTTPTransport
query1_server_answer_data = (
    '{"continents":['
    '{"code":"AF","name":"Africa"},{"code":"AN","name":"Antarctica"},'
    '{"code":"AS","name":"Asia"},{"code":"EU","name":"Europe"},'
    '{"code":"NA","name":"North America"},{"code":"OC","name":"Oceania"},'
    '{"code":"SA","name":"South America"}]}'
)


query1_server_answer = f'{{"data":{query1_server_answer_data}}}'

# Marking all tests in this file with the aiohttp marker
pytestmark = pytest.mark.aiohttp


@pytest.mark.asyncio
async def test_aiohttp_query(event_loop, aiohttp_server):

    async def handler(request):
        return web.Response(text=query1_server_answer, content_type="application/json")

    app = web.Application()
    app.router.add_route("POST", "/", handler)
    server = await aiohttp_server(app)

    url = server.make_url("/")

    sample_transport = AIOHTTPTransport(url=url, timeout=10)

    async with Client(transport=sample_transport,) as session:

        query = gql(query1_str)

        # Execute query asynchronously
        result = await session.execute(query)

        continents = result["continents"]

        africa = continents[0]

        assert africa["code"] == "AF"


@pytest.mark.asyncio
async def test_aiohttp_ignore_backend_content_type(event_loop, aiohttp_server):

    async def handler(request):
        return web.Response(text=query1_server_answer, content_type="text/plain")

    app = web.Application()
    app.router.add_route("POST", "/", handler)
    server = await aiohttp_server(app)

    url = server.make_url("/")

    sample_transport = AIOHTTPTransport(url=url, timeout=10)

    async with Client(transport=sample_transport,) as session:

        query = gql(query1_str)

        result = await session.execute(query)

        continents = result["continents"]

        africa = continents[0]

        assert africa["code"] == "AF"


@pytest.mark.asyncio
async def test_aiohttp_cookies(event_loop, aiohttp_server):

    async def handler(request):
        assert "COOKIE" in request.headers
        assert "cookie1=val1" == request.headers["COOKIE"]

        return web.Response(text=query1_server_answer, content_type="application/json")

    app = web.Application()
    app.router.add_route("POST", "/", handler)
    server = await aiohttp_server(app)

    url = server.make_url("/")

    sample_transport = AIOHTTPTransport(url=url, cookies={"cookie1": "val1"})

    async with Client(transport=sample_transport,) as session:

        query = gql(query1_str)

        # Execute query asynchronously
        result = await session.execute(query)

        continents = result["continents"]

        africa = continents[0]

        assert africa["code"] == "AF"


@pytest.mark.asyncio
async def test_aiohttp_error_code_401(event_loop, aiohttp_server):

    async def handler(request):
        # Will generate http error code 401
        return web.Response(
            text='{"error":"Unauthorized","message":"401 Client Error: Unauthorized"}',
            content_type="application/json",
            status=401,
        )

    app = web.Application()
    app.router.add_route("POST", "/", handler)
    server = await aiohttp_server(app)

    url = server.make_url("/")

    sample_transport = AIOHTTPTransport(url=url)


    async with Client(transport=sample_transport,) as session: