1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
|
import asyncio
import logging
import backoff
from gql import Client, gql
from gql.transport.aiohttp import AIOHTTPTransport
logging.basicConfig(level=logging.INFO)
async def main():
# Note: this example used the test backend from
# https://github.com/slothmanxyz/typegraphql-ws-apollo
transport = AIOHTTPTransport(url="ws://localhost:5000/graphql")
client = Client(transport=transport)
retry_connect = backoff.on_exception(
backoff.expo,
Exception,
max_value=10,
jitter=None,
)
session = await client.connect_async(reconnecting=True, retry_connect=retry_connect)
num = 0
while True:
num += 1
# Execute single query
query = gql("mutation ($message: String!) {sendMessage(message: $message)}")
query.variable_values = {"message": f"test {num}"}
try:
result = await session.execute(query)
print(result)
except Exception as e:
print(f"Received exception {e}")
await asyncio.sleep(1)
asyncio.run(main())
|
