Skip to content

DiskCache

from pulsefire.caches import DiskCache

Bases: BaseCache

Disk Cache.

Requires diskcache installed. This cache lives on disk, meaning cheaper storage but slower access.

Example: 

DiskCache() # Cache on tmp
DiskCache("folder") # Cache on folder/

Parameters:

Name Type Description Default
directory str | None

Cache directory, uses tmp if None.

 None
shards int

Number of shards to distribute writes.

 8
serializer

Serializer package supporting loads and dumps.

 pickle
Source code in pulsefire/caches.py 
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
class DiskCache(BaseCache):
    """Disk Cache.

    Requires `diskcache` installed. This cache lives on disk, meaning cheaper storage but slower access.

    Example:
    ```python
    DiskCache() # Cache on tmp
    DiskCache("folder") # Cache on folder/
    ```

    Parameters:
        directory: Cache directory, uses tmp if None.
        shards: Number of shards to distribute writes.
        serializer: Serializer package supporting `loads` and `dumps`.
    """

    def __init__(self, directory: str | None = None, shards: int = 8, serializer=pickle) -> None:
        import diskcache
        self.directory = directory
        self.serializer = serializer
        self.cache = diskcache.FanoutCache(directory, shards)

    @sync_to_async()
    def get[T](self, key: str) -> T:
        value = self.cache.get(key)
        if value is None:
            raise KeyError(key)
        return self.serializer.loads(value)

    @sync_to_async()
    def set(self, key: str, value: Any, ttl: float) -> None:
        if ttl <= 0:
            return
        if math.isinf(ttl):
            ttl = None
        self.cache.set(key, self.serializer.dumps(value), ttl)

    @sync_to_async()
    def clear(self) -> None:
        self.cache.clear()
Attributes
directory instance-attribute 
directory = directory
serializer instance-attribute 
serializer = serializer
cache instance-attribute 
cache = FanoutCache(directory, shards)
Functions
__init__ 
__init__(
    directory: str | None = None, shards: int = 8, serializer=pickle
) -> None
Source code in pulsefire/caches.py 
93
94
95
96
97
def __init__(self, directory: str | None = None, shards: int = 8, serializer=pickle) -> None:
    import diskcache
    self.directory = directory
    self.serializer = serializer
    self.cache = diskcache.FanoutCache(directory, shards)
get 
get(key: str) -> T
Source code in pulsefire/caches.py 
 99
100
101
102
103
104
@sync_to_async()
def get[T](self, key: str) -> T:
    value = self.cache.get(key)
    if value is None:
        raise KeyError(key)
    return self.serializer.loads(value)
set 
set(key: str, value: Any, ttl: float) -> None
Source code in pulsefire/caches.py 
106
107
108
109
110
111
112
@sync_to_async()
def set(self, key: str, value: Any, ttl: float) -> None:
    if ttl <= 0:
        return
    if math.isinf(ttl):
        ttl = None
    self.cache.set(key, self.serializer.dumps(value), ttl)
clear 
clear() -> None
Source code in pulsefire/caches.py 
114
115
116
@sync_to_async()
def clear(self) -> None:
    self.cache.clear()