mirror of
https://github.com/norohind/SquadsActivityMonitor.git
synced 2026-11-08 01:22:22 +03:00
Add get_latest_leaderboard
This commit is contained in:
5 files changed
+164
-45
No files matched your search
@@ -212,3 +212,29 @@ class PostgresModel(AbstractModel):
|
||||
cache.set(cache_key, json.dumps(result))
|
||||
|
||||
return result
|
||||
|
||||
@errors_catcher
|
||||
def get_latest_leaderboard(self, platform: str, leaderboard_type: str) -> list[dict]:
|
||||
cache_key = f'latest_leaderboard_{platform}_{leaderboard_type}'
|
||||
cached_result: typing.Union[str, None] = cache.get(cache_key)
|
||||
|
||||
if cached_result is not None:
|
||||
logger.debug(f'Cached result for {cache_key}')
|
||||
return json.loads(cached_result)
|
||||
|
||||
logger.debug(f'Not cached result for {cache_key}')
|
||||
|
||||
with self.db.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cursor:
|
||||
cursor.execute(
|
||||
postgres_sql_requests.select_latest_leaderboard,
|
||||
{
|
||||
'platform': platform.upper(),
|
||||
'LB_type': leaderboard_type.lower()
|
||||
}
|
||||
)
|
||||
result: list[dict] = cursor.fetchall()
|
||||
|
||||
if not cache.disabled:
|
||||
cache.set(cache_key, json.dumps(result))
|
||||
|
||||
return result
|
||||
Reference in new issue
Block a user