mirror of
https://github.com/Jylpah/blitz-tools.git
synced 2026-10-21 18:27:05 +02:00
Rate throtling v2, removed WG.create(), fixed player_stats retrieval + cache
This commit is contained in:
Jylpah committed
1 parent
cd605348e9
commit
5e4b7d858c
1 file changed
+68
-105
+68
-105
@@ -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)
|
||||
|
||||
Reference in new issue
Block a user