diff --git a/blitzutils.py b/blitzutils.py index 2a2e410..c8c086d 100644 --- a/blitzutils.py +++ b/blitzutils.py @@ -31,39 +31,36 @@ os.umask(UMASK) #### Class ThrottledClientSession(aiohttp.ClientSession) ## ----------------------------------------------------------- - class ThrottledClientSession(aiohttp.ClientSession): """Rate-throttled client session class inherited from aiohttp.ClientSession)""" - MAX_STEPS = 10 + MIN_SLEEP = 0.1 - def __init__(self, rate_limit: int =None, *args,**kwargs) -> None: + def __init__(self, rate_limit: float =None, *args,**kwargs) -> None: super().__init__(*args,**kwargs) self.rate_limit = rate_limit self._fillerTask = None self._queue = None - if self.rate_limit != None: - if self.rate_limit < 1: - raise ValueError('rate_limit must be positive integer') - (increment, sleep) = self._get_rate_increment() - self._queue = asyncio.Queue(rate_limit) - self._fillerTask = asyncio.create_task(self._filler(increment, sleep)) + if rate_limit != None: + if rate_limit <= 0: + raise ValueError('rate_limit must be positive') + #(increment, sleep) = self._get_rate_increment() + self._queue = asyncio.Queue(min(2, int(rate_limit)+1)) + self._fillerTask = asyncio.create_task(self._filler(rate_limit)) self._start_time = None self._count = 0 - def _get_rate_increment(self) -> list: - if self.rate_limit <= self.MAX_STEPS: - return (1, 1/self.rate_limit) - else: - return (self.rate_limit // self.MAX_STEPS, 1/self.MAX_STEPS) - + def _get_sleep(self) -> list: + return max(1/self.rate_limit, self.MIN_SLEEP) + async def close(self) -> None: """Close rate-limiter's "bucket filler" task""" # DEBUG if self._start_time != None: duration = time.time() - self._start_time debug('Average WG API request rate: ' + '{:.1f}'.format(self._count / duration) + ' / sec') - self._fillerTask.cancel() + if self._fillerTask != None: + self._fillerTask.cancel() try: await asyncio.wait_for(self._fillerTask, timeout= 3) except asyncio.TimeoutError as err: @@ -71,25 +68,35 @@ class ThrottledClientSession(aiohttp.ClientSession): await super().close() - async def _filler(self, increment: int = 1, sleep: float = 1): + async def _filler(self, rate_limit: float = 1): """Filler task to fill the leaky bucket algo""" try: if self._queue == None: return + self.rate_limit = rate_limit + sleep = self._get_sleep() + debug('SLEEP: ' + str(sleep)) + updated_at = time.monotonic() fraction = 0 extra_increment = 0 + for i in range(0,self._queue.maxsize): + self._queue.put_nowait(i) while True: if not self._queue.full(): - if self.rate_limit > self.MAX_STEPS: - fraction += (self.rate_limit - increment / sleep) / self.MAX_STEPS - extra_increment = fraction // 1 - fraction -= extra_increment - items_2_add = int(min(self.rate_limit - self._queue.qsize(), increment + extra_increment)) + now = time.monotonic() + increment = rate_limit * (now - updated_at) + fraction += increment % 1 + extra_increment = fraction // 1 + items_2_add = int(min(self._queue.maxsize - self._queue.qsize(), int(increment) + extra_increment)) + fraction = fraction % 1 for i in range(0,items_2_add): - self._queue.put_nowait(i) + self._queue.put_nowait(i) + updated_at = now await asyncio.sleep(sleep) except asyncio.CancelledError: debug('Cancelled') + except Exception as err: + error(exception=err) async def _allow(self) -> None: @@ -109,51 +116,6 @@ class ThrottledClientSession(aiohttp.ClientSession): await self._allow() return await super()._request(*args,**kwargs) - -## ----------------------------------------------------------- -#### Class asyncThrottle -## ----------------------------------------------------------- - -# class asyncThrottle: -# def __init__(self, rate: int = 20): -# if rate < 1: -# error('Rate must be positive integer') -# self.rate = rate -# self.queue = asyncio.Queue(rate) -# self.fillerTask = asyncio.create_task(self.filler()) -# self.start_time = time.time() -# self.count = 0 - - -# async def close(self): -# if self.start_time != None: -# duration = time.time() - self.start_time -# debug('Average WG API request rate: ' + '{:.1f}'.format(self.count / duration) + ' / sec', force=True) -# self.fillerTask.cancel() -# try: -# await asyncio.wait_for(self.fillerTask, timeout= 3) -# except asyncio.TimeoutError as err: -# error(exception=err) - - -# async def filler(self): -# try: -# while True: -# if not self.queue.full(): -# items_2_add = self.rate - self.queue.qsize() -# for i in range(0,items_2_add): -# self.queue.put_nowait(i) -# self.count += 1 -# await asyncio.sleep(1) -# except asyncio.CancelledError: -# debug('Cancelled') - -# async def allow(self) -> None: -# await self.queue.get() -# self.queue.task_done() -# return None - - def set_debug(debug: bool): global _log_level _log_level = DEBUG @@ -193,37 +155,39 @@ def get_log_level_str() -> str: error('Unknown log level: ' + str(_log_level)) -def verbose(msg = "", id = None): +def verbose(msg = "", id = None) -> bool: """Print a message""" if _log_level >= VERBOSE: if id == None: print(msg) else: print('[' + str(id) + ']: ' + msg) - return None + return True + return False -def verbose_std(msg = "", id = None): +def verbose_std(msg = "", id = None) -> bool: """Print a message""" if _log_level >= NORMAL: if id == None: print(msg) else: print('[' + str(id) + ']: ' + msg) - return None + return True + return False - -def debug(msg = "", id = None, exception = None, force: bool = False): +def debug(msg = "", id = None, exception = None, force: bool = False) -> bool: """print a conditional debug message""" if (_log_level >= DEBUG) or force: _print_log_msg('DEBUG', msg, exception, id) - return None + return True + return False -def error(msg = "", exception = None, id = None): +def error(msg = "", exception = None, id = None) -> bool: """Print an error message""" _print_log_msg('ERROR', msg, exception, id) - return None + return True def _print_log_msg(prefix = 'LOG', msg = '', exception = None, id = None): @@ -625,34 +589,18 @@ class WG: self.session = None debug('WG aiohttp session NOT initiated') - # cache TBD + # cache self.cache = None self.statsQ = None self.stat_saver_task = None - - - @classmethod - async def create(cls, WG_app_id = None, tankopedia_fn = None, maps_fn = None, stats_cache = False): - """Separete Constuctor method to handle async calls required to initialize the object""" - self = WG(WG_app_id, tankopedia_fn , maps_fn ) - if stats_cache: try: - self.cache = await aiosqlite.connect(WG.CACHE_DB_FILE) - ## Player Tank stats cache table - await self.cache.execute(WG.SQL_TANK_STATS_CREATE_TBL) - - await self.cache.commit() - - async with self.cache.execute(WG.SQL_TANK_STATS_COUNT) as cursor: - debug('Cache contains: ' + str((await cursor.fetchone())[0]) + ' cached player tank stat records' ) - self.statsQ = asyncio.Queue() self.stat_saver_task = asyncio.create_task(self.stat_saver()) except Exception as err: error(exception=err) sys.exit(1) - return self + async def close(self): # close stats queue @@ -879,6 +827,7 @@ class WG: error('Invalid tier', err) return None + def get_url_clan_info(self, server: str, clan_id: int) -> str: try: if server == None: @@ -988,7 +937,7 @@ class WG: #debug('JSON Response received: ' + str(json_data)) stats = json_data['data'][str(account_id)] if cache: - await self.save_stats('tank_stats', [account_id, tank_ids], stats) + await self.put_2_statsQ('tank_stats', [account_id, tank_ids], stats) return stats except Exception as err: error(exception=err) @@ -1020,7 +969,7 @@ class WG: #debug('JSON Response received: ' + str(json_data)) stats = json_data['data'][str(account_id)] if cache: - await self.save_stats('player_stats', [account_id], stats) + await self.put_2_statsQ('player_stats', [account_id], stats) return stats except Exception as err: error(exception=err) @@ -1050,12 +999,13 @@ class WG: except KeyError as err: error('Key not found', err) return None - + + def get_tank_tier(self, tank_id: int): return self.get_tank_data(tank_id, 'tier') - async def save_stats(self, statsType: str, key: list, stats: list): + async def put_2_statsQ(self, statsType: str, key: list, stats: list): """Save stats to a async queue to be saved by the stat_saver -task""" if self.statsQ == None: return False @@ -1063,12 +1013,26 @@ class WG: await self.statsQ.put([ statsType, key, stats, NOW() ]) return True + async def stat_saver(self): """Async task for saving stats into cache in background""" if self.statsQ == None: error('No statsQ defined') return None + try: + self.cache = await aiosqlite.connect(WG.CACHE_DB_FILE) + ## Create cache tables table + await self.cache.execute(WG.SQL_TANK_STATS_CREATE_TBL) + await self.cache.execute(WG.SQL_PLAYER_STATS_CREATE_TBL) + await self.cache.commit() + + async with self.cache.execute(WG.SQL_TANK_STATS_COUNT) as cursor: + debug('Cache contains: ' + str((await cursor.fetchone())[0]) + ' cached player tank stat records' ) + except Exception as err: + error(exception=err) + sys.exit(1) + while True: try: stats = await self.statsQ.get() @@ -1193,7 +1157,7 @@ class WG: #debug('No cache DB') raise StatsNotFound('No cache DB in use') - async with self.cache.execute(WG.SQL_PLAYER_STATS_CACHED, (account_id, NOW() - WG.CACHE_GRACE_TIME) ) as cursor: + async with self.cache.execute(WG.SQL_PLAYER_STATS_CACHED, [account_id, NOW() - WG.CACHE_GRACE_TIME] ) as cursor: row = await cursor.fetchone() #debug('account_id: ' + str(account_id) + ': 1') if row == None: @@ -1202,18 +1166,18 @@ class WG: raise StatsNotFound('No cached stats found') debug('Cached stats found') - if row[3] == None: + if row[2] == None: # None/null stats found in cache # i.e. stats have been requested, but not returned from WG API return None else: # Return proper stats - return json.loads(row[3]) + return json.loads(row[2]) except StatsNotFound as err: debug(exception=err) raise except Exception as err: - error('Error trying to look for cached stats', err) + error('Error trying to look for cached stats', exception=err) return None @@ -1334,8 +1298,7 @@ class WoTinspector: if json_resp.get('status', None) == None: error(msg_str +' : ' + title + ' : Received invalid JSON') elif (json_resp['status'] == 'ok'): - debug('Response data read') - verbose(msg_str + title + ' posted') + debug('Response data read. Status OK') return json_resp elif (json_resp['status'] == 'error'): error(msg_str + json_resp['error']['message'] + ' : ' + title)