44 lines
1.4 KiB
Python
44 lines
1.4 KiB
Python
from __future__ import annotations
|
|
|
|
from copy import deepcopy
|
|
from threading import Lock
|
|
from time import monotonic
|
|
from typing import Any
|
|
|
|
|
|
class PublicResponseCache:
|
|
def __init__(self, max_entries: int = 128) -> None:
|
|
self._items: dict[tuple[Any, ...], tuple[float, Any]] = {}
|
|
self._lock = Lock()
|
|
self._max_entries = max_entries
|
|
self._generation = 0
|
|
|
|
def generation(self) -> int:
|
|
with self._lock:
|
|
return self._generation
|
|
|
|
def get(self, key: tuple[Any, ...], ttl_seconds: int) -> Any | None:
|
|
with self._lock:
|
|
item = self._items.get(key)
|
|
if item is None or monotonic() - item[0] >= ttl_seconds:
|
|
self._items.pop(key, None)
|
|
return None
|
|
return deepcopy(item[1])
|
|
|
|
def set(self, key: tuple[Any, ...], value: Any, *, generation: int | None = None) -> Any:
|
|
with self._lock:
|
|
if generation is not None and generation != self._generation:
|
|
return value
|
|
if key not in self._items and len(self._items) >= self._max_entries:
|
|
self._items.pop(next(iter(self._items)))
|
|
self._items[key] = (monotonic(), deepcopy(value))
|
|
return value
|
|
|
|
def invalidate(self) -> None:
|
|
with self._lock:
|
|
self._generation += 1
|
|
self._items.clear()
|
|
|
|
|
|
public_cache = PublicResponseCache()
|