Spaces:
Sleeping
Sleeping
| import threading | |
| import time | |
| from collections import OrderedDict | |
| from typing import Optional | |
| class InMemoryCache: | |
| """ | |
| A simple in-memory cache using an OrderedDict. | |
| This cache supports setting a maximum size and expiration time for cached items. | |
| When the cache is full, it uses a Least Recently Used (LRU) eviction policy. | |
| Thread-safe using a threading Lock. | |
| Attributes: | |
| max_size (int, optional): Maximum number of items to store in the cache. | |
| expiration_time (int, optional): Time in seconds after which a cached item expires. Default is 1 hour. | |
| Example: | |
| cache = InMemoryCache(max_size=3, expiration_time=5) | |
| # setting cache values | |
| cache.set("a", 1) | |
| cache.set("b", 2) | |
| cache["c"] = 3 | |
| # getting cache values | |
| a = cache.get("a") | |
| b = cache["b"] | |
| """ | |
| def __init__(self, max_size: Optional[int] = None, expiration_time: Optional[int] = 60 * 60): | |
| """ | |
| Initialize a new InMemoryCache instance. | |
| Args: | |
| max_size (int, optional): Maximum number of items to store in the cache. | |
| expiration_time (int, optional): Time in seconds after which a cached item expires. Default is 1 hour. | |
| """ | |
| self._cache = OrderedDict() | |
| self._lock = threading.Lock() | |
| self.max_size = max_size | |
| self.expiration_time = expiration_time | |
| def get(self, key): | |
| """ | |
| Retrieve an item from the cache. | |
| Args: | |
| key: The key of the item to retrieve. | |
| Returns: | |
| The value associated with the key, or None if the key is not found or the item has expired. | |
| """ | |
| with self._lock: | |
| if key in self._cache: | |
| item = self._cache.pop(key) | |
| if ( | |
| self.expiration_time is None | |
| or time.time() - item["time"] < self.expiration_time | |
| ): | |
| # Move the key to the end to make it recently used | |
| self._cache[key] = item | |
| return item["value"] | |
| else: | |
| self.delete(key) | |
| return None | |
| def set(self, key, value): | |
| """ | |
| Add an item to the cache. | |
| If the cache is full, the least recently used item is evicted. | |
| Args: | |
| key: The key of the item. | |
| value: The value to cache. | |
| """ | |
| with self._lock: | |
| if key in self._cache: | |
| # Remove existing key before re-inserting to update order | |
| self.delete(key) | |
| elif self.max_size and len(self._cache) >= self.max_size: | |
| # Remove least recently used item | |
| self._cache.popitem(last=False) | |
| self._cache[key] = {"value": value, "time": time.time()} | |
| def get_or_set(self, key, value): | |
| """ | |
| Retrieve an item from the cache. If the item does not exist, set it with the provided value. | |
| Args: | |
| key: The key of the item. | |
| value: The value to cache if the item doesn't exist. | |
| Returns: | |
| The cached value associated with the key. | |
| """ | |
| with self._lock: | |
| if key in self._cache: | |
| return self.get(key) | |
| self.set(key, value) | |
| return value | |
| def delete(self, key): | |
| """ | |
| Remove an item from the cache. | |
| Args: | |
| key: The key of the item to remove. | |
| """ | |
| # with self._lock: | |
| self._cache.pop(key, None) | |
| def clear(self): | |
| """ | |
| Clear all items from the cache. | |
| """ | |
| with self._lock: | |
| self._cache.clear() | |
| def __contains__(self, key): | |
| """Check if the key is in the cache.""" | |
| return key in self._cache | |
| def __getitem__(self, key): | |
| """Retrieve an item from the cache using the square bracket notation.""" | |
| return self.get(key) | |
| def __setitem__(self, key, value): | |
| """Add an item to the cache using the square bracket notation.""" | |
| self.set(key, value) | |
| def __delitem__(self, key): | |
| """Remove an item from the cache using the square bracket notation.""" | |
| self.delete(key) | |
| def __len__(self): | |
| """Return the number of items in the cache.""" | |
| return len(self._cache) | |
| def __repr__(self): | |
| """Return a string representation of the InMemoryCache instance.""" | |
| return f"InMemoryCache(max_size={self.max_size}, expiration_time={self.expiration_time})" | |