gavindiaz/PolyWeather
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Persist scan terminal cache in Redis

Browse Source
This commit is contained in:
2569718930@qq.com
2026-06-01 01:53:54 +08:00
parent de5083ef18
commit 9a0d7ae444
3 changed files with 195 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docker-compose.yml
+2
View File
@@ -28,6 +28,7 @@ services:
env_file: &id001
- .env
environment:
POLYWEATHER_REDIS_URL: ${POLYWEATHER_REDIS_URL:-redis://polyweather_redis:6379/0}
POLYWEATHER_SCAN_TERMINAL_PREWARM_ENABLED: 'false'
POLYWEATHER_SERVICE_ROLE: bot
TELEGRAM_AIRPORT_PUSH_INTERVAL_SEC: ${POLYWEATHER_BOT_AIRPORT_PUSH_INTERVAL_SEC:-180}
@@ -96,6 +97,7 @@ services:
container_name: polyweather_web
env_file: *id001
environment:
POLYWEATHER_REDIS_URL: ${POLYWEATHER_REDIS_URL:-redis://polyweather_redis:6379/0}
POLYWEATHER_SERVICE_ROLE: web
healthcheck:
interval: 30s
tests/test_scan_terminal_modules.py
+60
View File
@@ -1,4 +1,5 @@
from web.scan_terminal_filters import normalize_scan_terminal_filters
from web import scan_terminal_cache
from web.scan_terminal_metar_gate import _apply_metar_gate_to_row
from web.scan_terminal_payloads import (
build_failed_scan_terminal_payload,
@@ -10,6 +11,65 @@ from web.scan_terminal_city_row import _build_quick_row
from web.routers.scan import router as scan_router
class _FakeRedis:
def __init__(self):
self.data = {}
def get(self, key):
return self.data.get(key)
def setex(self, key, _ttl, value):
self.data[key] = value
def test_scan_terminal_cache_hydrates_success_payload_from_redis(monkeypatch):
fake_redis = _FakeRedis()
monkeypatch.setenv("POLYWEATHER_SCAN_TERMINAL_REDIS_CACHE_ENABLED", "true")
monkeypatch.setattr(scan_terminal_cache, "_get_redis_client", lambda: fake_redis)
scan_terminal_cache._SCAN_TERMINAL_CACHE.clear()
filters = {"scan_mode": "tradable", "limit": 9}
payload = {
"generated_at": "2026-06-01T00:00:00Z",
"rows": [{"id": "row-1"}],
"summary": {"candidate_total": 1},
}
scan_terminal_cache.set_cached_scan_terminal_payload(filters, payload)
scan_terminal_cache._SCAN_TERMINAL_CACHE.clear()
entry = scan_terminal_cache.get_scan_terminal_cache_entry(filters)
cached = scan_terminal_cache.get_cached_scan_terminal_payload(filters, ttl_sec=3600)
assert entry["success_payload"]["rows"] == [{"id": "row-1"}]
assert cached["summary"]["candidate_total"] == 1
def test_scan_terminal_failure_state_preserves_redis_success_payload(monkeypatch):
fake_redis = _FakeRedis()
monkeypatch.setenv("POLYWEATHER_SCAN_TERMINAL_REDIS_CACHE_ENABLED", "true")
monkeypatch.setattr(scan_terminal_cache, "_get_redis_client", lambda: fake_redis)
scan_terminal_cache._SCAN_TERMINAL_CACHE.clear()
filters = {"scan_mode": "tradable", "limit": 9}
scan_terminal_cache.set_cached_scan_terminal_payload(
filters,
{
"generated_at": "2026-06-01T00:00:00Z",
"rows": [{"id": "row-1"}],
},
)
scan_terminal_cache._SCAN_TERMINAL_CACHE.clear()
scan_terminal_cache.set_scan_terminal_failure_state(filters, error_message="timeout")
scan_terminal_cache._SCAN_TERMINAL_CACHE.clear()
entry = scan_terminal_cache.get_scan_terminal_cache_entry(filters)
assert entry["success_payload"]["rows"] == [{"id": "row-1"}]
assert entry["last_error"] == "timeout"
def test_scan_router_does_not_expose_terminal_ai_endpoint():
routes = {
getattr(route, "path", None): getattr(route, "methods", set())
web/scan_terminal_cache.py
+133 -27
View File
@@ -1,6 +1,8 @@
from __future__ import annotations
import json
import hashlib
import os
import threading
import time
from datetime import datetime
@@ -9,39 +11,140 @@ from typing import Any, Dict, Optional
_SCAN_TERMINAL_CACHE_LOCK = threading.Lock()
_SCAN_TERMINAL_CACHE: Dict[str, Dict[str, Any]] = {}
_SCAN_TERMINAL_REFRESHING: set[str] = set()
_SCAN_TERMINAL_REDIS_CLIENT_LOCK = threading.Lock()
_SCAN_TERMINAL_REDIS_CLIENT: Any = None
_SCAN_TERMINAL_REDIS_UNAVAILABLE = False
def scan_terminal_cache_key(filters: Dict[str, Any]) -> str:
return json.dumps(filters, ensure_ascii=True, sort_keys=True)
def _truthy_env(name: str, *, default: bool = False) -> bool:
value = os.getenv(name)
if value is None:
return default
return str(value).strip().lower() not in ("", "0", "false", "no", "off")
def _redis_cache_enabled() -> bool:
return _truthy_env(
"POLYWEATHER_SCAN_TERMINAL_REDIS_CACHE_ENABLED",
default=bool(os.getenv("POLYWEATHER_REDIS_URL")),
)
def _redis_cache_ttl_sec() -> int:
try:
value = int(os.getenv("POLYWEATHER_SCAN_TERMINAL_REDIS_CACHE_TTL_SEC", "21600"))
except Exception:
value = 21600
return max(600, min(value, 86400))
def _redis_cache_prefix() -> str:
return os.getenv(
"POLYWEATHER_SCAN_TERMINAL_REDIS_CACHE_PREFIX",
"polyweather:scan_terminal:v1:",
)
def _redis_entry_key(cache_key: str) -> str:
digest = hashlib.sha256(cache_key.encode("utf-8")).hexdigest()
return f"{_redis_cache_prefix()}{digest}"
def _get_redis_client() -> Any:
global _SCAN_TERMINAL_REDIS_CLIENT, _SCAN_TERMINAL_REDIS_UNAVAILABLE
if not _redis_cache_enabled() or _SCAN_TERMINAL_REDIS_UNAVAILABLE:
return None
with _SCAN_TERMINAL_REDIS_CLIENT_LOCK:
if _SCAN_TERMINAL_REDIS_CLIENT is not None:
return _SCAN_TERMINAL_REDIS_CLIENT
try:
import redis # type: ignore
url = os.getenv("POLYWEATHER_REDIS_URL") or "redis://127.0.0.1:6379/0"
client = redis.Redis.from_url(
url,
socket_timeout=float(os.getenv("POLYWEATHER_REDIS_SOCKET_TIMEOUT_SECONDS", "2")),
socket_connect_timeout=float(
os.getenv("POLYWEATHER_REDIS_SOCKET_CONNECT_TIMEOUT_SECONDS", "1")
),
health_check_interval=30,
)
client.ping()
_SCAN_TERMINAL_REDIS_CLIENT = client
return client
except Exception:
_SCAN_TERMINAL_REDIS_UNAVAILABLE = True
return None
def _read_redis_cache_entry(cache_key: str) -> Optional[Dict[str, Any]]:
client = _get_redis_client()
if client is None:
return None
try:
raw = client.get(_redis_entry_key(cache_key))
if not raw:
return None
if isinstance(raw, bytes):
raw = raw.decode("utf-8")
entry = json.loads(str(raw))
return dict(entry) if isinstance(entry, dict) else None
except Exception:
return None
def _write_redis_cache_entry(cache_key: str, entry: Dict[str, Any]) -> None:
client = _get_redis_client()
if client is None:
return
try:
client.setex(
_redis_entry_key(cache_key),
_redis_cache_ttl_sec(),
json.dumps(entry, ensure_ascii=False, separators=(",", ":")),
)
except Exception:
return
def get_cached_scan_terminal_payload(
filters: Dict[str, Any],
*,
ttl_sec: int,
) -> Optional[Dict[str, Any]]:
cache_key = scan_terminal_cache_key(filters)
now = time.time()
with _SCAN_TERMINAL_CACHE_LOCK:
cached = _SCAN_TERMINAL_CACHE.get(cache_key)
if not cached:
return None
cached_at = float(cached.get("t") or 0.0)
if now - cached_at >= float(ttl_sec):
return None
payload = cached.get("payload")
if not isinstance(payload, dict):
return None
return dict(payload)
cached = get_scan_terminal_cache_entry(filters)
if not cached:
return None
cached_at = float(cached.get("t") or 0.0)
if now - cached_at >= float(ttl_sec):
return None
payload = cached.get("payload")
if not isinstance(payload, dict):
return None
return dict(payload)
def get_scan_terminal_cache_entry(filters: Dict[str, Any]) -> Optional[Dict[str, Any]]:
cache_key = scan_terminal_cache_key(filters)
with _SCAN_TERMINAL_CACHE_LOCK:
cached = _SCAN_TERMINAL_CACHE.get(cache_key)
if not isinstance(cached, dict):
return None
return dict(cached)
if isinstance(cached, dict):
return dict(cached)
redis_entry = _read_redis_cache_entry(cache_key)
if not redis_entry:
return None
with _SCAN_TERMINAL_CACHE_LOCK:
_SCAN_TERMINAL_CACHE[cache_key] = dict(redis_entry)
return dict(redis_entry)
def set_cached_scan_terminal_payload(
@@ -51,15 +154,17 @@ def set_cached_scan_terminal_payload(
cache_key = scan_terminal_cache_key(filters)
existing = get_scan_terminal_cache_entry(filters) or {}
now = time.time()
entry = {
"t": now,
"payload": dict(payload),
"success_t": now,
"success_payload": dict(payload),
"last_error": existing.get("last_error"),
"last_failed_at": existing.get("last_failed_at"),
}
with _SCAN_TERMINAL_CACHE_LOCK:
_SCAN_TERMINAL_CACHE[cache_key] = {
"t": now,
"payload": dict(payload),
"success_t": now,
"success_payload": dict(payload),
"last_error": existing.get("last_error"),
"last_failed_at": existing.get("last_failed_at"),
}
_SCAN_TERMINAL_CACHE[cache_key] = dict(entry)
_write_redis_cache_entry(cache_key, entry)
def set_scan_terminal_failure_state(
@@ -68,11 +173,12 @@ def set_scan_terminal_failure_state(
error_message: str,
) -> None:
cache_key = scan_terminal_cache_key(filters)
existing = get_scan_terminal_cache_entry(filters) or {}
existing["last_error"] = error_message
existing["last_failed_at"] = datetime.utcnow().isoformat() + "Z"
with _SCAN_TERMINAL_CACHE_LOCK:
existing = _SCAN_TERMINAL_CACHE.get(cache_key) or {}
existing["last_error"] = error_message
existing["last_failed_at"] = datetime.utcnow().isoformat() + "Z"
_SCAN_TERMINAL_CACHE[cache_key] = existing
_SCAN_TERMINAL_CACHE[cache_key] = dict(existing)
_write_redis_cache_entry(cache_key, existing)
def mark_scan_terminal_refreshing(filters: Dict[str, Any]) -> bool: