Spaces:
Runtime error
Runtime error
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) | |