Skip to content

API

Clients

The client implementation for aiohttp

AIOHTTPClient

Bases: AsyncClient

The aiohttp implementation for qlient.

Examples:

Basic Example

>>> async with AIOHTTPClient("https://...") as client:
>>>     result = await client.query.get_books(...)

With custom client session

>>> import aiohttp
>>> async with aiohttp.ClientSession() as session:
>>>     async with AIOHTTPClient(AIOHTTPBackend("https://...", session=session)) as client:
>>>         result = await client.query.get_books(...)
Source code in qlient/aiohttp/clients.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
class AIOHTTPClient(AsyncClient):
    """The aiohttp implementation for qlient.

    Examples:
        Basic Example
        >>> async with AIOHTTPClient("https://...") as client:
        >>>     result = await client.query.get_books(...)

        With custom client session
        >>> import aiohttp
        >>> async with aiohttp.ClientSession() as session:
        >>>     async with AIOHTTPClient(AIOHTTPBackend("https://...", session=session)) as client:
        >>>         result = await client.query.get_books(...)
    """

    def __init__(self, backend: Union[str, Backend], **kwargs):
        if isinstance(backend, str):
            backend = AIOHTTPBackend(backend)

        super(AIOHTTPClient, self).__init__(backend, **kwargs)

Backends

AIOHTTPBackend

Bases: AsyncBackend

Source code in qlient/aiohttp/backends.py 
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
class AIOHTTPBackend(AsyncBackend):
    @classmethod
    def generate_subscription_id(cls) -> str:
        """Class method to generate unique subscription ids

        Returns:
            A unique subscription id
        """
        return f"qlient:{cls.__name__}:{uuid.uuid4()}".replace("-", "")

    @staticmethod
    def adapt_to_websocket_endpoint(endpoint: str) -> str:
        """Adapt the http endpoint to websocket endpoint

        Args:
            endpoint: the endpoint

        Returns:
            a websocket url
        """
        if endpoint.startswith("https://"):
            return "wss://" + endpoint.replace("https://", "")
        elif endpoint.startswith("http://"):
            return "ws://" + endpoint.replace("http://", "")
        else:
            return endpoint

    @staticmethod
    def make_payload(request: GraphQLRequest) -> Dict[str, Any]:
        """Static method for generating the request payload

        Args:
            request: holds the graphql request

        Returns:
            the payload to send as dictionary
        """
        return {
            "query": request.query,
            "operationName": request.operation_name,
            "variables": request.variables,
        }

    def __init__(
        self,
        endpoint: str,
        ws_endpoint: Optional[str] = None,
        session: Optional[aiohttp.ClientSession] = None,
        subscription_protocols: Optional[List[str]] = None,
    ):
        if ws_endpoint is None:
            ws_endpoint = AIOHTTPBackend.adapt_to_websocket_endpoint(endpoint)

        if not subscription_protocols:
            subscription_protocols = [
                GRAPHQL_WS_PROTOCOL,
                GRAPHQL_TRANSPORT_WS_PROTOCOL,
            ]

        self.endpoint: str = endpoint
        self.ws_endpoint: str = ws_endpoint
        self.subscription_protocols = subscription_protocols
        self._session: Optional[aiohttp.ClientSession] = session

    @property
    @asynccontextmanager
    async def session(self) -> aiohttp.ClientSession:
        if self._session is not None:
            yield self._session
            return

        async with aiohttp.ClientSession() as session:
            yield session

    async def execute_query(self, request: GraphQLRequest) -> GraphQLResponse:
        """

        Args:
            request:

        Returns:

        """
        payload = self.make_payload(request)
        async with self.session as session:
            async with session.post(self.endpoint, json=payload) as response:
                response_body = await response.json()
                return GraphQLResponse(request, response_body)

    async def execute_mutation(self, request: GraphQLRequest) -> GraphQLResponse:
        """

        Args:
            request:

        Returns:

        """
        return await self.execute_query(request)

    async def execute_subscription(
        self, request: GraphQLSubscriptionRequest
    ) -> GraphQLResponse:
        """

        Args:
            request:

        Returns:

        """
        payload = self.make_payload(request)
        async with self.session as session:
            request.subscription_id = (
                request.subscription_id or self.generate_subscription_id()
            )
            ws = await session.ws_connect(
                self.endpoint, protocols=self.subscription_protocols, autoclose=False
            )
            # initiate connection
            await ws.send_json({"type": CONNECTION_INIT, "payload": request.options})

            initial_response = await ws.receive_json()
            if initial_response.get("type") != CONNECTION_ACKNOWLEDGED:
                logger.critical(f"The server did not acknowledged the connection.")
                raise ConnectionRejected(
                    "The server did not acknowledge the connection."
                )

            # connection acknowledged, send request
            await ws.send_json(
                {"type": START, "id": request.subscription_id, "payload": payload}
            )

            SUBSCRIPTION_ID_TO_WS[request.subscription_id] = ws

            async def _make_generator() -> AsyncGenerator:
                msg: aiohttp.WSMessage
                async for msg in ws:
                    if msg.type == aiohttp.WSMsgType.ERROR:
                        # break the iterator
                        await ws.close()
                        break
                    if msg.type != aiohttp.WSMsgType.TEXT:
                        raise TypeError(
                            f"Expected {aiohttp.WSMsgType.TEXT}; Got {msg.type}"
                        )

                    data = msg.json()
                    data_type = data["type"]

                    if data_type in (CONNECTION_TERMINATE, CONNECTION_ERROR, COMPLETE):
                        # break the iterator
                        await ws.close()
                        break

                    if data_type == CONNECTION_KEEP_ALIVE:
                        continue

                    yield GraphQLResponse(request, data["payload"])

            return GraphQLResponse(request, _make_generator())

adapt_to_websocket_endpoint(endpoint) staticmethod

Adapt the http endpoint to websocket endpoint

Parameters:

Name Type Description Default
endpoint str

the endpoint

 required

Returns:

Type Description
str

a websocket url
Source code in qlient/aiohttp/backends.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
@staticmethod
def adapt_to_websocket_endpoint(endpoint: str) -> str:
    """Adapt the http endpoint to websocket endpoint

    Args:
        endpoint: the endpoint

    Returns:
        a websocket url
    """
    if endpoint.startswith("https://"):
        return "wss://" + endpoint.replace("https://", "")
    elif endpoint.startswith("http://"):
        return "ws://" + endpoint.replace("http://", "")
    else:
        return endpoint

execute_mutation(request) async

Parameters:

Name Type Description Default
request GraphQLRequest required
Source code in qlient/aiohttp/backends.py 
127
128
129
130
131
132
133
134
135
136
async def execute_mutation(self, request: GraphQLRequest) -> GraphQLResponse:
    """

    Args:
        request:

    Returns:

    """
    return await self.execute_query(request)

execute_query(request) async

Parameters:

Name Type Description Default
request GraphQLRequest required
Source code in qlient/aiohttp/backends.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
async def execute_query(self, request: GraphQLRequest) -> GraphQLResponse:
    """

    Args:
        request:

    Returns:

    """
    payload = self.make_payload(request)
    async with self.session as session:
        async with session.post(self.endpoint, json=payload) as response:
            response_body = await response.json()
            return GraphQLResponse(request, response_body)

execute_subscription(request) async

Parameters:

Name Type Description Default
request GraphQLSubscriptionRequest required
Source code in qlient/aiohttp/backends.py 
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
async def execute_subscription(
    self, request: GraphQLSubscriptionRequest
) -> GraphQLResponse:
    """

    Args:
        request:

    Returns:

    """
    payload = self.make_payload(request)
    async with self.session as session:
        request.subscription_id = (
            request.subscription_id or self.generate_subscription_id()
        )
        ws = await session.ws_connect(
            self.endpoint, protocols=self.subscription_protocols, autoclose=False
        )
        # initiate connection
        await ws.send_json({"type": CONNECTION_INIT, "payload": request.options})

        initial_response = await ws.receive_json()
        if initial_response.get("type") != CONNECTION_ACKNOWLEDGED:
            logger.critical(f"The server did not acknowledged the connection.")
            raise ConnectionRejected(
                "The server did not acknowledge the connection."
            )

        # connection acknowledged, send request
        await ws.send_json(
            {"type": START, "id": request.subscription_id, "payload": payload}
        )

        SUBSCRIPTION_ID_TO_WS[request.subscription_id] = ws

        async def _make_generator() -> AsyncGenerator:
            msg: aiohttp.WSMessage
            async for msg in ws:
                if msg.type == aiohttp.WSMsgType.ERROR:
                    # break the iterator
                    await ws.close()
                    break
                if msg.type != aiohttp.WSMsgType.TEXT:
                    raise TypeError(
                        f"Expected {aiohttp.WSMsgType.TEXT}; Got {msg.type}"
                    )

                data = msg.json()
                data_type = data["type"]

                if data_type in (CONNECTION_TERMINATE, CONNECTION_ERROR, COMPLETE):
                    # break the iterator
                    await ws.close()
                    break

                if data_type == CONNECTION_KEEP_ALIVE:
                    continue

                yield GraphQLResponse(request, data["payload"])

        return GraphQLResponse(request, _make_generator())

generate_subscription_id() classmethod

Class method to generate unique subscription ids

Returns:

Type Description
str

A unique subscription id
Source code in qlient/aiohttp/backends.py 
39
40
41
42
43
44
45
46
@classmethod
def generate_subscription_id(cls) -> str:
    """Class method to generate unique subscription ids

    Returns:
        A unique subscription id
    """
    return f"qlient:{cls.__name__}:{uuid.uuid4()}".replace("-", "")

make_payload(request) staticmethod

Static method for generating the request payload

Parameters:

Name Type Description Default
request GraphQLRequest

holds the graphql request

 required

Returns:

Type Description
Dict[str, Any]

the payload to send as dictionary
Source code in qlient/aiohttp/backends.py 
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
@staticmethod
def make_payload(request: GraphQLRequest) -> Dict[str, Any]:
    """Static method for generating the request payload

    Args:
        request: holds the graphql request

    Returns:
        the payload to send as dictionary
    """
    return {
        "query": request.query,
        "operationName": request.operation_name,
        "variables": request.variables,
    }