Diskcache - Fast Local Cache

Keywords: Cache

Reference:

• Quick Start

• Tutorial

Create Cache

import diskcache

import os
import time
from pathlib import Path

dir_here = Path(os.getcwd())
dir_cache = dir_here / ".cache"
cache = diskcache.Cache(str(dir_cache))

Common Operation

• contains, key in cache: Return True if key matching item is found in cache.

• set(key, value, expire=None, read=False, tag=None, retry=False): Set key and value item in cache.

• get(key, default=None, read=False, expire_time=False, tag=False, retry=False): Retrieve value from cache. If key is missing, return default.

• delete(key, retry=False): Delete corresponding item for key from cache.

• add(key, value, expire=None, read=False, tag=None, retry=False):

• pop(key, default=None, expire_time=False, tag=False, retry=False): Remove corresponding item for key from cache and return value. If key is missing, return default.

• touch(key, expire=None, retry=False): Touch key in cache and update expire time.

• clear(retry=False): Remove all items from cache.

• expire(now=None, retry=False): Remove expired items from cache.

• evict(tag, retry=False): Remove items with matching tag from cache.

• create_tag_index(): Create tag index on cache database.

• drop_tag_index(): Drop tag index on cache database.

• memoize(name=None, typed=False, expire=None, tag=None, ignore=()): Memoizing cache decorator.

• read(key, retry=False): Return file handle value corresponding to key from cache.

• incr(key, delta=1, default=0, retry=False): Increment value by delta for item with key.

• decr(key, delta=1, default=0, retry=False): Decrement value by delta for item with key.

• iterkeys(reverse=False): Iterate Cache keys in database sort order.

• peek(prefix=None, default=(None, None), side=‘front’, expire_time=False, tag=False, retry=False): Peek at key and value item pair from side of queue in cache.

• peekitem(last=True, expire_time=False, tag=False, retry=False): Peek at key and value item pair in cache based on iteration order.

• push(value, prefix=None, side=‘back’, expire=None, read=False, tag=None, retry=False): Push value onto side of queue identified by prefix in cache.

• pull(prefix=None, default=(None, None), side=‘front’, expire_time=False, tag=False, retry=False): Pull key and value item pair from side of queue in cache.

• transact(retry=False): Context manager to perform a transaction by locking the cache.

• stats(enable=True, reset=False): Return cache statistics hits and misses.

• volume(): Return estimated total size of cache on disk.

• check(fix=False, retry=False)¶: Check database and file system consistency. Intended for use in testing and post-mortem error analysis.

• reset(key, value=ENOVAL, update=True): Reset key and value item from Settings table.

• cull(retry=False): Cull items from cache until volume is less than size limit.

Basic

cache.set("my-key", "my-value")
cache.get("my-key")

'my-value'

"my-key" in cache

True

print(cache.get("unknown-key"))

None

"unknown-key" in cache

False

Cache with expire

key = "my-key"
cache.set(key, "my-value", expire=1)
print(f"first time: {cache.get(key)}")
time.sleep(1.5)
print(f"second time: {cache.get(key)}")

first time: my-value
second time: None

Cache with Tag

Tag is the metadata of a cached item. Tag value can be int, float, string, bytes or None. Tag can be used to

key = "my-key"
cache.set(key, "my-value", tag="my-tag")
cache.get(key)

'my-value'

cache.evict("my-tag")
print(cache.get(key))

None

Memoize

class CloudData:
    def __init__(self, backend_id: str):
        self.backend_id = backend_id

    def read_data(self):
        print("reading data from cloud")
        return [
            {"name": "alice"},
            {"name": "bob"},
            {"name": "cathy"},
        ]

    @cache.memoize(expire=1)
    def query_data(self, name):
        data = self.read_data()
        return [d for d in data if d["name"] == name]

cloud_data_1 = CloudData("backend-1")

print("query 1")
print(cloud_data_1.query_data(name="alice"))

print("query 2")
print(cloud_data_1.query_data(name="alice"))

time.sleep(1.5)
print("query 3")
print(cloud_data_1.query_data(name="alice"))

query 1
reading data from cloud
[{'name': 'alice'}]
query 2
[{'name': 'alice'}]
query 3
reading data from cloud
[{'name': 'alice'}]

Evict

cache.clear()
tag = "my-tag"
cache.set("k1", 1, tag=tag)
cache.set("k2", 1, tag=tag)
cache.set("k3", 1, tag=tag)

print("k1, k2, k3 is in cache")
print(cache.get("k1"))
print(cache.get("k2"))
print(cache.get("k3"))

cache.evict(tag)
print("k1, k2, k3 is not in cache")
print(cache.get("k1"))
print(cache.get("k2"))
print(cache.get("k3"))

k1, k2, k3 is in cache
1
1
1
k1, k2, k3 is not in cache
None
None
None