Source code for aries_cloudagent.cache.basic

"""Basic in-memory cache implementation."""

import time
from typing import Any, Sequence, Text, Union

from .base import BaseCache

[docs]class BasicCache(BaseCache): """Basic in-memory cache class.""" def __init__(self): """Initialize a `BasicCache` instance.""" super().__init__() # looks like { "key": { "expires": <epoch timestamp>, "value": <val> } } self._cache = {} def _remove_expired_cache_items(self): """Remove all expired items from cache.""" for key, val in self._cache.copy().items(): # iterate copy, del from original cache_item_expiry = val["expires"] if cache_item_expiry is None: continue now = time.perf_counter() if now >= cache_item_expiry: del self._cache[key]
[docs] async def get(self, key: Text): """ Get an item from the cache. Args: key: the key to retrieve an item for Returns: The record found or `None` """ self._remove_expired_cache_items() return self._cache.get(key)["value"] if self._cache.get(key) else None
[docs] async def set(self, keys: Union[Text, Sequence[Text]], value: Any, ttl: int = None): """ Add an item to the cache with an optional ttl. Overwrites existing cache entries. Args: keys: the key or keys for which to set an item value: the value to store in the cache ttl: number of seconds that the record should persist """ self._remove_expired_cache_items() expires_ts = time.perf_counter() + ttl if ttl else None for key in [keys] if isinstance(keys, Text) else keys: self._cache[key] = {"expires": expires_ts, "value": value}
[docs] async def clear(self, key: Text): """ Remove an item from the cache, if present. Args: key: the key to remove """ if key in self._cache: del self._cache[key]
[docs] async def flush(self): """Remove all items from the cache.""" self._cache = {}