Skip to content

MemoryCache

from pulsefire.caches import MemoryCache

Bases: BaseCache

Memory Cache.

This cache lives in-memory, be aware of memory footprint when caching large responses.

Example: 

MemoryCache()

Source code in pulsefire/caches.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
class MemoryCache(BaseCache):
    """Memory Cache.

    This cache lives in-memory, be aware of memory footprint when caching large responses.

    Example:
    ```python
    MemoryCache()
    ```
    """

    cache: dict[str, tuple[Any, float]]

    def __init__(self) -> None:
        self.cache = {}
        self.last_expired = time.time()

    async def get[T](self, key: str) -> T:
        value, expire = self.cache[key]
        if time.time() > expire:
            self.cache.pop(key, None)
            raise KeyError(key)
        return value

    async def set(self, key: str, value: Any, ttl: float) -> None:
        if ttl <= 0:
            return
        self.cache[key] = [value, time.time() + ttl]
        if time.time() - self.last_expired > 60:
            self.last_expired = time.time()
            for old_key, (_, expire) in self.cache.items():
                if time.time() > expire:
                    self.cache.pop(old_key, None)

    async def clear(self) -> None:
        self.cache.clear()
Attributes
cache instance-attribute 
cache: dict[str, tuple[Any, float]] = {}
last_expired instance-attribute 
last_expired = time()
Functions
__init__ 
__init__() -> None
Source code in pulsefire/caches.py 
51
52
53
def __init__(self) -> None:
    self.cache = {}
    self.last_expired = time.time()
get async 
get(key: str) -> T
Source code in pulsefire/caches.py 
55
56
57
58
59
60
async def get[T](self, key: str) -> T:
    value, expire = self.cache[key]
    if time.time() > expire:
        self.cache.pop(key, None)
        raise KeyError(key)
    return value
set async 
set(key: str, value: Any, ttl: float) -> None
Source code in pulsefire/caches.py 
62
63
64
65
66
67
68
69
70
async def set(self, key: str, value: Any, ttl: float) -> None:
    if ttl <= 0:
        return
    self.cache[key] = [value, time.time() + ttl]
    if time.time() - self.last_expired > 60:
        self.last_expired = time.time()
        for old_key, (_, expire) in self.cache.items():
            if time.time() > expire:
                self.cache.pop(old_key, None)
clear async 
clear() -> None
Source code in pulsefire/caches.py 
72
73
async def clear(self) -> None:
    self.cache.clear()