"""简单的 TTL 内存缓存(纯 Python,无第三方依赖)。 对应 TS SDK 的 MemoryCache,用于主数据等就近缓存,生命周期跟随 client 实例。 注意:这是客户端层缓存,与 Prefect 的 task 缓存(cache_key_fn)是两回事。 """ from __future__ import annotations import time from typing import Any, Generic, TypeVar T = TypeVar("T") _DEFAULT_TTL_SECONDS = 300 class TTLCache(Generic[T]): def __init__(self, ttl_seconds: int | float = _DEFAULT_TTL_SECONDS) -> None: self._ttl = ttl_seconds self._store: dict[str, tuple[T, float]] = {} def get(self, key: str) -> T | None: item = self._store.get(key) if item is None: return None value, expires_at = item if time.monotonic() > expires_at: self._store.pop(key, None) return None return value def set(self, key: str, value: T) -> None: self._store[key] = (value, time.monotonic() + self._ttl) def clear(self) -> None: self._store.clear()