database.py (1467B)
1 import asyncio 2 from typing import Dict, List 3 4 import aiohttp 5 6 from token_bot.token_database.exceptions import * 7 from token_bot.token_database.flavor import Flavor 8 from token_bot.token_database.region import Region 9 10 11 class Database: 12 def __init__(self, session: aiohttp.ClientSession): 13 self.data_url = "https://data.wowtoken.app/v2/" 14 self.session = session 15 16 async def _get_data(self, endpoint: str) -> Dict | List: 17 url = f"{self.data_url}{endpoint}" 18 19 success = False 20 tries = 0 21 backoff = 0.1 22 23 while not success and tries <= 5: 24 if tries > 1: 25 await asyncio.sleep(backoff * tries) 26 try: 27 return await self._external_call(url) 28 except TokenHttpException: 29 tries += 1 30 raise TokenHttpException 31 32 async def _external_call(self, url) -> Dict | List: 33 async with self.session.get(url) as resp: 34 await resp.text() 35 if resp.ok: 36 return await resp.json() 37 else: 38 raise TokenHttpException(resp.status) 39 40 async def current(self, flavor: Flavor) -> dict: 41 return await self._get_data(f"current/{flavor.name.lower()}.json") 42 43 async def history(self, flavor: Flavor, region: Region, relative_time: str = "all"): 44 return await self._get_data( 45 f"relative/{flavor.name.lower()}/{region.value.lower()}/{relative_time}.json" 46 )