Intermediate

In-memory caching with functools.lru_cache is fast and easy, but it evaporates the moment your process ends. Every time your script restarts, every time your web server worker recycles, the cache is empty and you rebuild everything from scratch. For expensive operations — API calls with rate limits, slow database queries, heavy computations, or file parsing that takes several seconds — losing the cache on every restart is a real cost. You also cannot share an in-memory cache between separate processes, which is a problem in multi-worker web servers and parallel batch jobs.

diskcache is a Python caching library that persists data to disk using SQLite, so your cache survives process restarts and is shareable between multiple processes. The API is designed to feel like a Python dictionary — you set values, get values, check membership, and set expiration times. It also provides a @cache.memoize() decorator that drops directly on top of existing functions, and a FanoutCache for high-concurrency scenarios. Install it with pip install diskcache.

This article covers creating a cache, basic CRUD operations, setting time-to-live expiration, using the memoize decorator, handling cache size limits, and a realistic example showing a disk-cached API client that avoids redundant requests across multiple script runs. By the end, you will have a persistent caching layer ready for any Python project that benefits from reusing expensive results.

Quick Example: Persistent Key-Value Cache

Here is a minimal demonstration. The key detail is that after the first run, subsequent runs read from disk instead of recomputing.

# quick_diskcache.py
import diskcache
import time

# Cache is stored in the ./my_cache directory on disk
cache = diskcache.Cache("./my_cache")

def slow_operation(key):
    """Simulates an expensive computation (e.g., API call, DB query)."""
    time.sleep(1)
    return f"result_for_{key}"

# First access -- computes and stores
if "item_a" not in cache:
    print("Computing item_a...")
    cache["item_a"] = slow_operation("item_a")

# Second access -- reads from disk (instant)
print("Reading item_a:", cache["item_a"])

cache.close()
print("Cache closed -- value persists on disk")

First run output:

Computing item_a...
Reading item_a: result_for_item_a
Cache closed -- value persists on disk

Second run output (instant, no sleep):

Reading item_a: result_for_item_a
Cache closed -- value persists on disk

The value persists between runs because it is stored in SQLite files under ./my_cache/. The in operator checks for key existence just like a dict, and dictionary-style assignment stores the value. Run the script twice to see the difference — the second run has no sleep because the value was already on disk.

What Is diskcache and When Should You Use It?

diskcache stores cache entries in a SQLite database (for metadata and small values) and flat files on disk (for large binary values). It handles concurrent access from multiple processes safely, supports atomic operations, and respects configurable size limits. The library was designed to replace Redis or Memcached in situations where you want persistence without the overhead of running a separate cache server.

OptionPersists?Multi-process?DependenciesBest For
lru_cacheNoNoNone (stdlib)In-process function memoization
diskcacheYesYesNone (no server)Persistent caching without infrastructure
RedisYesYesRedis server requiredDistributed caching, pub/sub
shelveYesNoNone (stdlib)Simple persistent dict, single process

diskcache is the right choice when you want persistence and multi-process access without running a Redis server. It is particularly useful for development environments, batch scripts, data pipelines, and single-machine web applications with multiple worker processes.

Basic Cache Operations

diskcache’s Cache object supports dictionary-style access, method-based access, and several cache-specific operations including TTL expiration and tagging.

# cache_operations.py
import diskcache

cache = diskcache.Cache("./demo_cache")

# Set values (dict-style or set() method)
cache["username"] = "alice"
cache.set("score", 99)

# Set with time-to-live (TTL) in seconds
cache.set("session_token", "abc123xyz", expire=60)  # expires in 60 seconds

# Get values
print(cache["username"])        # alice
print(cache.get("score"))       # 99
print(cache.get("missing_key")) # None (no KeyError)
print(cache.get("missing_key", default="fallback"))  # fallback

# Check existence
print("username" in cache)      # True
print("ghost" in cache)         # False

# Delete a key
del cache["username"]
print("username" in cache)      # False

# Atomic increment (thread/process safe)
cache.set("page_views", 0)
cache.incr("page_views")
cache.incr("page_views")
cache.incr("page_views")
print("Page views:", cache.get("page_views"))  # 3

# Get cache stats
print("Cache size (items):", len(cache))
print("Cache size (bytes):", cache.volume())

cache.close()

Output:

alice
99
None
fallback
True
False
False
Page views: 3
Cache size (items): 3
Cache size (bytes): 8192

The incr() method is especially useful for counters shared between processes — it uses SQLite’s atomic increment so concurrent processes do not clobber each other’s counts. The volume() method returns total disk usage in bytes, which lets you monitor cache growth.

The memoize() Decorator

The most convenient diskcache feature for most applications is the @cache.memoize() decorator. It caches the return value of a function keyed by its arguments, with an optional TTL. The result persists across process restarts.

# memoize_demo.py
import diskcache
import time
import requests

cache = diskcache.Cache("./api_cache")

@cache.memoize(expire=3600)  # cache results for 1 hour
def fetch_user(user_id: int) -> dict:
    """Fetches user data from a REST API -- cached for 1 hour."""
    print(f"  [CACHE MISS] Fetching user {user_id} from API...")
    response = requests.get(f"https://jsonplaceholder.typicode.com/users/{user_id}")
    return response.json()

@cache.memoize(expire=300)   # cache for 5 minutes
def fetch_posts_for_user(user_id: int) -> list:
    """Fetches all posts by a user -- cached for 5 minutes."""
    print(f"  [CACHE MISS] Fetching posts for user {user_id}...")
    response = requests.get(f"https://jsonplaceholder.typicode.com/posts?userId={user_id}")
    return response.json()

print("=== First run (populates cache) ===")
start = time.time()
user = fetch_user(1)
posts = fetch_posts_for_user(1)
elapsed = time.time() - start
print(f"User: {user['name']}, Posts: {len(posts)}, Time: {elapsed:.2f}s")

print("\n=== Second run (reads from disk cache) ===")
start = time.time()
user = fetch_user(1)
posts = fetch_posts_for_user(1)
elapsed = time.time() - start
print(f"User: {user['name']}, Posts: {len(posts)}, Time: {elapsed:.2f}s")

cache.close()

Output (first run):

=== First run (populates cache) ===
  [CACHE MISS] Fetching user 1 from API...
  [CACHE MISS] Fetching posts for user 1...
User: Leanne Graham, Posts: 10, Time: 0.84s

=== Second run (reads from disk cache) ===
User: Leanne Graham, Posts: 10, Time: 0.002s

The second run took 2 milliseconds instead of 840 milliseconds because both results were already on disk. On the next script run (a completely new process), the cache will still be there and the first call will also be fast — unlike lru_cache which would start fresh.

Cache Size Limits and Eviction

diskcache lets you set a maximum cache size. When the cache exceeds that size, it evicts the least recently used entries to stay within the limit.

# cache_size.py
import diskcache

# Limit cache to 10 MB
cache = diskcache.Cache("./bounded_cache", size_limit=10 * 1024 * 1024)

# Store some values
for i in range(100):
    cache.set(f"key_{i}", "x" * 10000)  # ~10KB per entry

print("Items in cache:", len(cache))
print("Volume (bytes):", cache.volume())
print("Size limit (bytes):", cache.size_limit)

# Manually evict entries if needed
evicted = cache.evict(tag=None)  # evict all entries without a tag

# Clear the entire cache
cache.clear()
print("After clear, items:", len(cache))

cache.close()

Output:

Items in cache: 100
Volume (bytes): 1245184
Size limit (bytes): 10485760
After clear, items: 0

Eviction happens automatically — you do not need to call any cleanup function. diskcache tracks access times and removes old entries as new ones are written. Setting a reasonable size limit (e.g., 500MB for a data pipeline, 50MB for a web server) prevents the cache directory from growing indefinitely.

Real-Life Example: Cached Data Enrichment Pipeline

Here is a realistic data enrichment pipeline that fetches additional details for a list of records. Without caching, every run makes the same API calls. With diskcache, only the first run pays the network cost.

# enrichment_pipeline.py
import diskcache
import requests
import time
from dataclasses import dataclass
from typing import Optional

CACHE = diskcache.Cache("./enrichment_cache", size_limit=100 * 1024 * 1024)

@dataclass
class EnrichedUser:
    id: int
    name: str
    email: str
    company: str
    website: Optional[str]

@CACHE.memoize(expire=86400)  # cache for 24 hours
def fetch_user_details(user_id: int) -> dict:
    """Fetches user details from REST API. Expensive -- cache aggressively."""
    resp = requests.get(f"https://jsonplaceholder.typicode.com/users/{user_id}", timeout=10)
    resp.raise_for_status()
    return resp.json()

def enrich_users(user_ids: list) -> list:
    """Enrich a list of user IDs with full profile data."""
    results = []
    cache_hits = 0

    for uid in user_ids:
        # Check if already cached before calling memoized function
        key = ("fetch_user_details", uid)
        was_cached = key in CACHE

        raw = fetch_user_details(uid)
        if was_cached:
            cache_hits += 1

        user = EnrichedUser(
            id=raw["id"],
            name=raw["name"],
            email=raw["email"],
            company=raw["company"]["name"],
            website=raw.get("website"),
        )
        results.append(user)

    print(f"Processed {len(user_ids)} users ({cache_hits} from cache, {len(user_ids)-cache_hits} from API)")
    return results

# Run the pipeline with user IDs 1-5
print("=== Pipeline Run ===")
start = time.time()
users = enrich_users([1, 2, 3, 4, 5])
elapsed = time.time() - start

for u in users:
    print(f"  {u.id}. {u.name} ({u.company}) -- {u.email}")
print(f"Total time: {elapsed:.2f}s")

CACHE.close()

Output (first run):

=== Pipeline Run ===
Processed 5 users (0 from cache, 5 from API)
  1. Leanne Graham (Romaguera-Crona) -- Sincere@april.biz
  2. Ervin Howell (Deckow-Crist) -- Shanna@melissa.tv
  ...
Total time: 1.23s

Output (subsequent runs — same day):

=== Pipeline Run ===
Processed 5 users (5 from cache, 0 from API)
  1. Leanne Graham (Romaguera-Crona) -- Sincere@april.biz
  ...
Total time: 0.01s

Subsequent runs are 100x faster because all 5 API calls hit the disk cache. The 24-hour TTL means the cache refreshes daily, so data does not go stale indefinitely. You can extend this pattern to cache database query results, file parsing output, or any other expensive step in a pipeline.

Frequently Asked Questions

Is diskcache safe for concurrent access?

Yes. diskcache uses SQLite’s WAL (Write-Ahead Logging) mode, which supports concurrent reads and serialized writes. Multiple threads and multiple processes can read from and write to the same cache directory simultaneously without corrupting data. For extremely high concurrency (hundreds of writes per second), use FanoutCache instead, which shards the data across multiple SQLite files to reduce lock contention.

How does diskcache differ from Python’s shelve module?

shelve is a stdlib persistent dict that stores data in a DBM file. It does not support concurrent process access, does not have TTL expiration, and does not enforce size limits. diskcache supports all of these plus atomic increment, memoize decorators, cache statistics, and FanoutCache for high concurrency. Use shelve only for simple single-process persistence; use diskcache whenever you need expiration, size limits, or multi-process access.

How does TTL expiration work?

Expired entries are not immediately deleted — they are filtered out on read. When you call cache.get(key), diskcache checks the entry’s expiration time and returns None (or the default) if it has expired, even if the entry still exists on disk. Expired entries are cleaned up during the next cache culling cycle (triggered automatically when the cache exceeds its size limit, or manually via cache.expire()). This means disk usage can temporarily exceed your expectations if you have many short-lived entries — call cache.expire() periodically to clean up proactively.

Where is the cache stored and can I delete it?

The cache is stored in the directory path you pass to Cache(). It consists of SQLite files (cache.db) and flat files for large values. You can delete the entire directory to clear the cache completely. There is no background process or daemon — the cache directory is just files that Python reads and writes. This makes it trivial to deploy (no Redis setup needed) and trivial to reset (just delete the folder).

Can I store custom Python objects?

Yes. diskcache uses pickle for serialization by default, so any picklable Python object can be stored — dataclasses, custom classes, numpy arrays, PIL images, pandas DataFrames. If you need cross-language compatibility or need to store JSON specifically, pass disk=diskcache.JSONDisk when creating the cache. The JSONDisk variant stores values as JSON text instead of pickle, which is human-readable in the SQLite file but limited to JSON-serializable types.

Conclusion

diskcache closes the gap between in-memory caching (fast but ephemeral) and a full Redis deployment (persistent but requires infrastructure). With a dictionary-style API, automatic TTL expiration, process-safe concurrent access, and a memoize decorator that drops onto existing functions, it handles the majority of persistent caching needs in a single pip install diskcache. The size limit and eviction system keeps disk usage bounded, and the SQLite storage makes the cache inspectable and trivially deletable.

Use the memoize decorator for API calls and expensive computations, set TTLs based on how stale your data can reasonably be, and configure a size limit to prevent unbounded growth. For high-write-concurrency scenarios, swap Cache for FanoutCache with the same API. The full reference — including FanoutCache, DjangoCache integration, and Deque/Index data structures built on the same storage — is available at the official diskcache documentation.