37 lines
1.0 KiB
Python
37 lines
1.0 KiB
Python
"""简单的 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()
|