72 lines
2.0 KiB
Python
72 lines
2.0 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from threading import RLock
|
|
import time
|
|
from typing import Callable, TypeVar
|
|
|
|
T = TypeVar("T")
|
|
|
|
|
|
@dataclass
|
|
class _CacheEntry:
|
|
value: object
|
|
expires_at: float
|
|
revision: int
|
|
|
|
|
|
class RuntimeCache:
|
|
"""Simple in-memory cache for local/prototype use.
|
|
|
|
Cache is globally invalidated by `bump_revision()` which we call after CUD.
|
|
"""
|
|
|
|
def __init__(self) -> None:
|
|
self._lock = RLock()
|
|
self._revision = 0
|
|
self._entries: dict[str, _CacheEntry] = {}
|
|
|
|
def get(self, key: str) -> object | None:
|
|
now = time.time()
|
|
with self._lock:
|
|
entry = self._entries.get(key)
|
|
if not entry:
|
|
return None
|
|
if entry.expires_at <= now or entry.revision != self._revision:
|
|
self._entries.pop(key, None)
|
|
return None
|
|
return entry.value
|
|
|
|
def set(self, key: str, value: object, ttl_seconds: int = 30) -> object:
|
|
now = time.time()
|
|
with self._lock:
|
|
self._entries[key] = _CacheEntry(
|
|
value=value,
|
|
expires_at=now + max(ttl_seconds, 1),
|
|
revision=self._revision,
|
|
)
|
|
if len(self._entries) > 2000:
|
|
self._entries.clear()
|
|
return value
|
|
|
|
def get_or_set(self, key: str, factory: Callable[[], T], ttl_seconds: int = 30) -> T:
|
|
cached = self.get(key)
|
|
if cached is not None:
|
|
return cached # type: ignore[return-value]
|
|
return self.set(key, factory(), ttl_seconds=ttl_seconds) # type: ignore[return-value]
|
|
|
|
def bump_revision(self) -> int:
|
|
with self._lock:
|
|
self._revision += 1
|
|
if self._revision > 1_000_000_000:
|
|
self._revision = 1
|
|
self._entries.clear()
|
|
return self._revision
|
|
|
|
def revision(self) -> int:
|
|
with self._lock:
|
|
return self._revision
|
|
|
|
|
|
runtime_cache = RuntimeCache()
|