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
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
|
import asyncio
import logging
from typing import Optional
from aioconsole import ainput
from gql import Client, gql
from gql.client import AsyncClientSession
from gql.transport.aiohttp import AIOHTTPTransport
logging.basicConfig(level=logging.INFO)
GET_CONTINENT_NAME = """
query getContinentName ($code: ID!) {
continent (code: $code) {
name
}
}
"""
class GraphQLContinentClient:
def __init__(self):
self._client = Client(
transport=AIOHTTPTransport(url="https://countries.trevorblades.com/")
)
self._session: Optional[AsyncClientSession] = None
self.get_continent_name_query = gql(GET_CONTINENT_NAME)
async def connect(self):
self._session = await self._client.connect_async(reconnecting=True)
async def close(self):
await self._client.close_async()
async def get_continent_name(self, code):
self.get_continent_name_query.variable_values = {"code": code}
assert self._session is not None
answer = await self._session.execute(self.get_continent_name_query)
return answer.get("continent").get("name") # type: ignore
async def main():
continent_client = GraphQLContinentClient()
continent_codes = ["AF", "AN", "AS", "EU", "NA", "OC", "SA"]
await continent_client.connect()
while True:
answer = await ainput("\nPlease enter a continent code or 'exit':")
answer = answer.strip()
if answer == "exit":
break
elif answer in continent_codes:
try:
continent_name = await continent_client.get_continent_name(answer)
print(f"The continent name is {continent_name}\n")
except Exception as exc:
print(f"Received exception {exc} while trying to get continent name")
else:
print(f"Please enter a valid continent code from {continent_codes}")
await continent_client.close()
asyncio.run(main())
|
