From 87f9e23bae37db23e0089256b052df4a3c79df2d Mon Sep 17 00:00:00 2001 From: 3252a8 <3252a8@proton.me> Date: Sat, 16 May 2026 00:01:40 +0300 Subject: [PATCH] refactor: cache panel squad and host lookups in-memory with TTL --- bot/services/panel_api_service.py | 31 +++++++++++++++++ bot/utils/ttl_cache.py | 55 +++++++++++++++++++++++++++++++ 2 files changed, 86 insertions(+) create mode 100644 bot/utils/ttl_cache.py diff --git a/bot/services/panel_api_service.py b/bot/services/panel_api_service.py index ee7cbb2..d450c89 100644 --- a/bot/services/panel_api_service.py +++ b/bot/services/panel_api_service.py @@ -9,6 +9,7 @@ from urllib.parse import urlencode import aiohttp from sqlalchemy.ext.asyncio import AsyncSession +from bot.utils.ttl_cache import AsyncTTLCache from config.settings import Settings from db.dal import panel_sync_dal from db.models import PanelSyncStatus @@ -27,6 +28,10 @@ class PanelApiService: self.api_key = settings.PANEL_API_KEY self._session: Optional[aiohttp.ClientSession] = None self.default_client_ip = "127.0.0.1" + # Cache slow-changing reference data fetched from the panel. Errors and + # None responses are not cached, so transient failures self-heal. + self._squads_cache: AsyncTTLCache = AsyncTTLCache(ttl_seconds=300) + self._hosts_cache: AsyncTTLCache = AsyncTTLCache(ttl_seconds=300) async def __aenter__(self): """Context manager entry""" @@ -604,7 +609,13 @@ class PanelApiService: ) return None + def _invalidate_squad_caches(self) -> None: + self._squads_cache.invalidate() + async def get_internal_squads(self) -> Optional[List[Dict[str, Any]]]: + return await self._squads_cache.get_or_load("list", self._get_internal_squads_uncached) + + async def _get_internal_squads_uncached(self) -> Optional[List[Dict[str, Any]]]: response_data = await self._request("GET", "/internal-squads", log_full_response=False) if response_data and not response_data.get("error") and "response" in response_data: response = response_data.get("response") @@ -619,6 +630,12 @@ class PanelApiService: return None async def get_internal_squad(self, squad_uuid: str) -> Optional[Dict[str, Any]]: + return await self._squads_cache.get_or_load( + f"detail:{squad_uuid}", + lambda: self._get_internal_squad_uncached(squad_uuid), + ) + + async def _get_internal_squad_uncached(self, squad_uuid: str) -> Optional[Dict[str, Any]]: response_data = await self._request( "GET", f"/internal-squads/{squad_uuid}", log_full_response=False ) @@ -639,6 +656,15 @@ class PanelApiService: async def get_internal_squad_accessible_nodes( self, squad_uuid: str, + ) -> Optional[List[Dict[str, Any]]]: + return await self._squads_cache.get_or_load( + f"nodes:{squad_uuid}", + lambda: self._get_internal_squad_accessible_nodes_uncached(squad_uuid), + ) + + async def _get_internal_squad_accessible_nodes_uncached( + self, + squad_uuid: str, ) -> Optional[List[Dict[str, Any]]]: endpoints = ( f"/internal-squads/{squad_uuid}/accessible-nodes", @@ -665,6 +691,9 @@ class PanelApiService: return None async def get_hosts(self) -> Optional[List[Dict[str, Any]]]: + return await self._hosts_cache.get_or_load("list", self._get_hosts_uncached) + + async def _get_hosts_uncached(self) -> Optional[List[Dict[str, Any]]]: response_data = await self._request("GET", "/hosts", log_full_response=False) if response_data and not response_data.get("error") and "response" in response_data: response = response_data.get("response") @@ -695,6 +724,7 @@ class PanelApiService: log_full_response=False, ) if response_data and not response_data.get("error"): + self._invalidate_squad_caches() return True logging.error("Failed to add users to squad %s. Response: %s", squad_uuid, response_data) return False @@ -710,6 +740,7 @@ class PanelApiService: log_full_response=False, ) if response_data and not response_data.get("error"): + self._invalidate_squad_caches() return True logging.error( "Failed to remove users from squad %s. Response: %s", squad_uuid, response_data diff --git a/bot/utils/ttl_cache.py b/bot/utils/ttl_cache.py new file mode 100644 index 0000000..bf5bd1f --- /dev/null +++ b/bot/utils/ttl_cache.py @@ -0,0 +1,55 @@ +import asyncio +import time +from typing import Any, Awaitable, Callable, Dict, Optional, Tuple + + +class AsyncTTLCache: + """In-memory async-safe TTL cache with single-flight loader. + + Concurrent get_or_load() calls for the same key share one loader execution. + """ + + def __init__(self, ttl_seconds: float): + self.ttl_seconds = ttl_seconds + self._data: Dict[str, Tuple[float, Any]] = {} + self._locks: Dict[str, asyncio.Lock] = {} + + def _is_fresh(self, expires_at: float) -> bool: + return time.monotonic() < expires_at + + def get_fresh(self, key: str) -> Optional[Any]: + entry = self._data.get(key) + if entry is None: + return None + expires_at, value = entry + if not self._is_fresh(expires_at): + return None + return value + + @staticmethod + def _is_cacheable(value: Any) -> bool: + if value is None: + return False + if isinstance(value, dict) and value.get("error"): + return False + return True + + async def get_or_load(self, key: str, loader: Callable[[], Awaitable[Any]]) -> Any: + cached = self.get_fresh(key) + if cached is not None: + return cached + lock = self._locks.setdefault(key, asyncio.Lock()) + async with lock: + cached = self.get_fresh(key) + if cached is not None: + return cached + value = await loader() + if self._is_cacheable(value): + self._data[key] = (time.monotonic() + self.ttl_seconds, value) + return value + + def invalidate(self, key: Optional[str] = None) -> None: + if key is None: + self._data.clear() + return + self._data.pop(key, None)