33 lines
911 B
Python
33 lines
911 B
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) -> None:
|
|
self._items: dict[tuple[Any, ...], tuple[float, Any]] = {}
|
|
self._lock = Lock()
|
|
|
|
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) -> Any:
|
|
with self._lock:
|
|
self._items[key] = (monotonic(), deepcopy(value))
|
|
return value
|
|
|
|
def invalidate(self) -> None:
|
|
with self._lock:
|
|
self._items.clear()
|
|
|
|
|
|
public_cache = PublicResponseCache()
|