55 lines
1.7 KiB
Python
55 lines
1.7 KiB
Python
import logging
|
|
from typing import Any, Dict, Optional
|
|
|
|
from fastapi import Request
|
|
|
|
from core.cache import cache
|
|
|
|
logger = logging.getLogger(__name__)
|
|
PLATFORM_OVERVIEW_CACHE_KEY = "platform:overview"
|
|
PLATFORM_OVERVIEW_CACHE_TTL_SECONDS = 15
|
|
PLATFORM_NODES_CACHE_KEY = "platform:nodes:list"
|
|
PLATFORM_NODES_CACHE_TTL_SECONDS = 20
|
|
|
|
|
|
def cached_platform_overview_payload() -> Optional[Dict[str, Any]]:
|
|
cached = cache.get_json(PLATFORM_OVERVIEW_CACHE_KEY)
|
|
return cached if isinstance(cached, dict) else None
|
|
|
|
|
|
def store_platform_overview_payload(payload: Dict[str, Any]) -> Dict[str, Any]:
|
|
cache.set_json(PLATFORM_OVERVIEW_CACHE_KEY, payload, ttl=PLATFORM_OVERVIEW_CACHE_TTL_SECONDS)
|
|
return payload
|
|
|
|
|
|
def invalidate_platform_overview_cache() -> None:
|
|
cache.delete(PLATFORM_OVERVIEW_CACHE_KEY)
|
|
|
|
|
|
def cached_platform_nodes_payload() -> Optional[Dict[str, Any]]:
|
|
cached = cache.get_json(PLATFORM_NODES_CACHE_KEY)
|
|
if not isinstance(cached, dict):
|
|
return None
|
|
items = cached.get("items")
|
|
if not isinstance(items, list):
|
|
return None
|
|
return {"items": items}
|
|
|
|
|
|
def store_platform_nodes_payload(items: list[Dict[str, Any]]) -> Dict[str, Any]:
|
|
payload = {"items": items}
|
|
cache.set_json(PLATFORM_NODES_CACHE_KEY, payload, ttl=PLATFORM_NODES_CACHE_TTL_SECONDS)
|
|
return payload
|
|
|
|
|
|
def invalidate_platform_nodes_cache() -> None:
|
|
cache.delete(PLATFORM_NODES_CACHE_KEY)
|
|
|
|
|
|
def apply_platform_runtime_changes(request: Request) -> None:
|
|
invalidate_platform_overview_cache()
|
|
invalidate_platform_nodes_cache()
|
|
speech_service = getattr(request.app.state, "speech_service", None)
|
|
if speech_service is not None and hasattr(speech_service, "reset_runtime"):
|
|
speech_service.reset_runtime()
|