Spaces:
Runtime error
Runtime error
File size: 811 Bytes
e00b837 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 |
import typing
import time
class CacheMixin:
# A simple in-memory cache with TTL (thread safe). This is used to cache tokenizers at the moment.
_cache: typing.Dict[str, typing.Tuple[typing.Optional[float], typing.Any]] = dict()
def _cache_get(self, key: str) -> typing.Any:
val = self._cache.get(key)
if val is None:
return None
expiry_timestamp, value = val
if expiry_timestamp is None or expiry_timestamp > time.time():
return value
del self._cache[key] # remove expired cache entry
def _cache_set(self, key: str, value: typing.Any, ttl: int = 60 * 60) -> None:
expiry_timestamp = None
if ttl is not None:
expiry_timestamp = time.time() + ttl
self._cache[key] = (expiry_timestamp, value)
|