Disk Persistence

async-cache supports optional disk-backed storage so your cache survives process restarts. This is critical for expensive ML inference caches, web scraper results, or any data that takes minutes to warm up.

No new dependencies — uses Python’s built-in pickle module.

Quick Start

from cache import AsyncCache, DiskBackend

backup = DiskBackend("/tmp/my_app_cache.pkl")
cache = AsyncCache(maxsize=10000, default_ttl=3600, backup=backup)

# Use normally — cache loads from disk on init
value = await cache.get("key", loader=lambda: expensive_computation())

# On shutdown, persist to disk
cache.save_to_backup()

How It Works

1. On init: AsyncCache calls backup.load() to warm up from disk

2. During runtime: All operations are in-memory (same performance)

3. On shutdown: Call save_to_backup() to persist current entries

4. TTL preservation: Remaining TTL is saved and restored correctly

import atexit
from cache import AsyncCache, DiskBackend

backup = DiskBackend("/var/cache/myapp/cache.pkl")
cache = AsyncCache(maxsize=5000, default_ttl=600, backup=backup)

# Auto-save on shutdown
atexit.register(cache.save_to_backup)

With Decorators

from cache import AsyncLRU, AsyncTTL, DiskBackend

lru_backup = DiskBackend("/tmp/product_cache.pkl")

@AsyncLRU(maxsize=1000, backup=lru_backup)
async def get_product(product_id: int):
    return await db.query_product(product_id)

# Save on shutdown
get_product.save_to_backup()

# Reload from disk manually
get_product.load_from_backup()

With AgentCache

from agent_cache import AgentCache
from cache import DiskBackend

@AgentCache(resource="embeddings", scope="global", backup=DiskBackend("/tmp/embeddings.pkl"))
async def get_embedding(text):
    return await ml_model.embed(text)

Custom Backends

To build a Redis or Memcached backend, subclass CacheBackend:

from cache.backend import CacheBackend

class RedisBackend(CacheBackend):
    def __init__(self, redis_url):
        self.redis = redis.from_url(redis_url)

    def load(self):
        # Return {key: (value, monotonic_expiration_or_None)}
        ...

    def save(self, data):
        # data is {key: (value, ttl_remaining_seconds_or_None)}
        ...

    def clear(self):
        self.redis.flushdb()

The CacheBackend ABC requires three methods:

Method	Description
load()	Return {key: (value, expiration)} dict. expiration is a monotonic deadline (float) or None.
save(data)	Persist {key: (value, ttl_remaining_or_None)} where ttl_remaining is seconds remaining (float) or None.
clear()	Remove all persisted data.

DiskBackend Details

• Atomic writes: Uses write-to-temp + os.replace() to prevent corruption

• Auto-creates directories: Parent directories are created if missing

• Corrupt file handling: Returns empty dict on unpickling errors

• Thread-safe: Internal locking for concurrent save/load

• Timestamp tracking: Elapsed time between save and load is subtracted from TTLs