diff --git a/quantara/web_app/api/leaderboard.py b/quantara/web_app/api/leaderboard.py index 4e43c509..0a5d5804 100644 --- a/quantara/web_app/api/leaderboard.py +++ b/quantara/web_app/api/leaderboard.py @@ -5,9 +5,15 @@ from web_app.db.crud.leaderboard import LeaderboardDBConnector from web_app.api.serializers.leaderboard import UserLeaderboardItem, TokenPositionStatistic from web_app.api.rate_limiter import limiter, READ_LIMIT +from web_app.contract_tools.cache import get_cached_or_fetch router = APIRouter() leaderboard_db_connector = LeaderboardDBConnector() +LEADERBOARD_CACHE_TTL_SECONDS = 30 +USER_LEADERBOARD_CACHE_NAME = ":".join(("leaderboard", "user", "top_positions")) +POSITION_TOKEN_STATISTICS_CACHE_NAME = ":".join( + ("leaderboard", "position_tokens", "statistics") +) @router.get( "/api/get-user-leaderboard", @@ -21,8 +27,14 @@ async def get_user_leaderboard(request: Request) -> list[UserLeaderboardItem]: """ Get the top 10 users ordered by closed/opened positions. """ - leaderboard_data = leaderboard_db_connector.get_top_users_by_positions() - return leaderboard_data + async def fetch_leaderboard(): + return leaderboard_db_connector.get_top_users_by_positions() + + return await get_cached_or_fetch( + USER_LEADERBOARD_CACHE_NAME, + LEADERBOARD_CACHE_TTL_SECONDS, + fetch_leaderboard, + ) @router.get( @@ -38,4 +50,11 @@ async def get_position_tokens_statistic(request: Request) -> list[TokenPositionS This endpoint retrieves statistics about positions grouped by token symbol. Returns counts of opened and closed positions for each token. """ - return leaderboard_db_connector.get_position_token_statistics() + async def fetch_token_statistics(): + return leaderboard_db_connector.get_position_token_statistics() + + return await get_cached_or_fetch( + POSITION_TOKEN_STATISTICS_CACHE_NAME, + LEADERBOARD_CACHE_TTL_SECONDS, + fetch_token_statistics, + ) diff --git a/quantara/web_app/contract_tools/cache.py b/quantara/web_app/contract_tools/cache.py index fefe6bed..ef51531f 100644 --- a/quantara/web_app/contract_tools/cache.py +++ b/quantara/web_app/contract_tools/cache.py @@ -4,6 +4,7 @@ import os from typing import Awaitable, Callable, Optional +import redis as redis_sync import redis.asyncio as redis logger = logging.getLogger(__name__) @@ -52,3 +53,21 @@ async def get_cached_or_fetch( return value finally: await client.close() + + +def delete_cache_pattern_sync(pattern: str) -> None: + """Best-effort synchronous cache invalidation for DB lifecycle hooks.""" + client = redis_sync.Redis.from_url(_REDIS_URL, decode_responses=True) + try: + keys = list(client.scan_iter(match=pattern)) + if keys: + client.delete(*keys) + except Exception as exc: + logger.warning("Cache invalidation failed for %s: %s", pattern, exc) + finally: + client.close() + + +def invalidate_leaderboard_cache() -> None: + """Clear cached leaderboard aggregates after position lifecycle changes.""" + delete_cache_pattern_sync("leaderboard:*") \ No newline at end of file diff --git a/quantara/web_app/db/crud/position.py b/quantara/web_app/db/crud/position.py index f7a85a4e..b97d5529 100644 --- a/quantara/web_app/db/crud/position.py +++ b/quantara/web_app/db/crud/position.py @@ -13,6 +13,7 @@ from sqlalchemy.dialects.postgresql import insert from sqlalchemy.exc import SQLAlchemyError +from web_app.contract_tools.cache import invalidate_leaderboard_cache from web_app.db.models import Base, ExtraDeposit, Position, Status, Transaction, User from .user import UserDBConnector @@ -281,6 +282,7 @@ def close_position(self, position_id: uuid) -> Position | None: position.status = Status.CLOSED.value position.closed_at = datetime.now() self.write_to_db(position) + invalidate_leaderboard_cache() return position.status def open_position(self, position_id: uuid.UUID, current_prices: dict) -> str | None: @@ -296,6 +298,7 @@ def open_position(self, position_id: uuid.UUID, current_prices: dict) -> str | N self.write_to_db(position) self.create_empty_claim(position.user_id) self.save_current_price(position, current_prices) + invalidate_leaderboard_cache() return position.status else: logger.error("db_open_position_not_found", position_id=str(position_id)) @@ -407,6 +410,7 @@ def liquidate_position(self, position_id: UUID) -> bool: self.write_to_db(position) logger.info("db_position_liquidated", position_id=str(position_id)) + invalidate_leaderboard_cache() return True except SQLAlchemyError as e: diff --git a/quantara/web_app/tests/test_leaderboard_cache_static.py b/quantara/web_app/tests/test_leaderboard_cache_static.py new file mode 100644 index 00000000..6b42d2a1 --- /dev/null +++ b/quantara/web_app/tests/test_leaderboard_cache_static.py @@ -0,0 +1,63 @@ +from pathlib import Path +import unittest + + +ROOT = Path(__file__).resolve().parents[1] +LEADERBOARD_API = ROOT / "api" / "leaderboard.py" +POSITION_CRUD = ROOT / "db" / "crud" / "position.py" +CACHE_HELPER = ROOT / "contract_tools" / "cache.py" + + +class LeaderboardCacheStaticTests(unittest.TestCase): + def test_leaderboard_endpoints_use_cache_names_and_ttl(self): + source = LEADERBOARD_API.read_text() + + self.assertIn( + "from web_app.contract_tools.cache import get_cached_or_fetch", + source, + ) + self.assertIn("LEADERBOARD_CACHE_TTL_SECONDS = 30", source) + self.assertIn("USER_LEADERBOARD_CACHE_NAME", source) + self.assertIn("POSITION_TOKEN_STATISTICS_CACHE_NAME", source) + self.assertIn( + '":".join(("leaderboard", "user", "top_positions"))', + source, + ) + self.assertIn( + '("leaderboard", "position_tokens", "statistics")', + source, + ) + self.assertIn("return await get_cached_or_fetch(", source) + self.assertIn( + "leaderboard_db_connector.get_top_users_by_positions()", + source, + ) + self.assertIn( + "leaderboard_db_connector.get_position_token_statistics()", + source, + ) + + def test_position_lifecycle_invalidates_leaderboard_cache(self): + source = POSITION_CRUD.read_text() + + self.assertIn( + "from web_app.contract_tools.cache import invalidate_leaderboard_cache", + source, + ) + self.assertGreaterEqual(source.count("invalidate_leaderboard_cache()"), 3) + self.assertIn("def close_position", source) + self.assertIn("def open_position", source) + self.assertIn("def liquidate_position", source) + + def test_cache_helper_has_sync_pattern_invalidation(self): + source = CACHE_HELPER.read_text() + + self.assertIn("import redis as redis_sync", source) + self.assertIn("def delete_cache_pattern_sync(pattern: str) -> None:", source) + self.assertIn("client.scan_iter(match=pattern)", source) + self.assertIn("client.delete(*keys)", source) + self.assertIn('delete_cache_pattern_sync("leaderboard:*")', source) + + +if __name__ == "__main__": + unittest.main()