Cachetta for Python

File-based caching for Python. Uses pickle for native binary serialization – any file extension works, and all picklable types (sets, tuples, bytes, dataclasses, etc.) are supported natively.

Install

uv add cachetta

Requires Python 3.12+.

Basic Usage

Create a cache object:

from datetime import timedelta
from cachetta import Cachetta

cache = Cachetta(
    read=True,
    write=True,
    path='./cache.json',
    duration=timedelta(days=1),
)

Read and write:

from cachetta import read_cache, write_cache

def get_data():
    with read_cache(cache) as cached_data:
        if cached_data:
            return cached_data
    data = fetch_data()
    write_cache(cache, data)
    return data

Decorators

Use Cachetta as a decorator:

from cachetta import Cachetta

@Cachetta(path='/my-cache.json')
def get_data():
    return compute_expensive_value()

With a specific cache object:

cache = Cachetta(path='/my-cache.json')

@cache
def get_data():
    return compute_expensive_value()

With overrides:

cache = Cachetta(path='/my-cache.json')

@cache(duration=timedelta(hours=1))
def get_data():
    return compute_expensive_value()

Async Support

Cachetta works seamlessly with async functions. When decorating an async function, all file I/O is automatically performed in background threads via asyncio.to_thread():

import asyncio
from cachetta import Cachetta

@Cachetta(path='./async-cache.json')
async def get_async_data():
    await asyncio.sleep(2)
    return {"status": "success", "data": [1, 2, 3]}

For explicit async cache operations:

from cachetta import async_read_cache, async_write_cache

async def get_data():
    async with async_read_cache(cache) as cached_data:
        if cached_data is not None:
            return cached_data
    data = await fetch_data()
    await async_write_cache(cache, data)
    return data

Function Wrapper

If you’re not using decorators, wrap functions manually:

cache = Cachetta(path='./my-cache.json')

def get_data():
    return compute_expensive_value()

cached_get_data = cache(get_data)
result = cached_get_data()

With configuration:

cache = Cachetta(path='./cache')

cached_get_data = cache(get_data, duration=timedelta(hours=2))
result = cached_get_data(123)

Auto Cache Keys

When a decorated function receives arguments, Cachetta automatically generates unique cache paths by hashing the arguments:

@Cachetta(path='./cache/users.json')
def get_user(user_id: int):
    return fetch_user(user_id)

get_user(1)   # cached at ./cache/users-<hash1>.json
get_user(2)   # cached at ./cache/users-<hash2>.json

In-Memory LRU

Add an in-memory LRU layer that is checked before hitting disk:

cache = Cachetta(
    path='./cache.json',
    lru_size=100,
)

LRU entries respect the same duration as disk entries. The LRU is thread-safe for concurrent async access.

Conditional Caching

Cache results only when a condition function returns True:

cache = Cachetta(
    path='./cache.json',
    condition=lambda result: result is not None,
)

Stale-While-Revalidate

Return expired data immediately while refreshing in the background:

cache = Cachetta(
    path='./cache.json',
    duration=timedelta(hours=1),
    stale_duration=timedelta(minutes=30),
)

Cache Invalidation

cache = Cachetta(path='./cache.json')

cache.invalidate()  # or cache.clear()

# With arguments
cache.invalidate(user_id=123)

# Async variants
await cache.ainvalidate()
await cache.aclear()

Cache Inspection

Query cache state without reading the cached data:

cache = Cachetta(path='./cache.json')

cache.exists()   # True if the cache file exists
cache.age()      # timedelta or None
cache.info()     # {"exists": True, "age": timedelta(...), "expired": False, ...}

# Async variants
await cache.aexists()
await cache.aage()
await cache.ainfo()

Path Operator

Use the / operator to build sub-paths:

cache = Cachetta(path='./cache')

with read_cache(cache / 'my-data.json') as data:
    ...

Dynamic Cache Paths

Specify a function for defining the path:

def get_cache_path(n: int):
    return f"./cache/{n}.json"

@Cachetta(path=get_cache_path)
def foo(n: int):
    return compute_expensive_value(n)

Specifying Paths

Use copy to create variations of a cache configuration:

cache = Cachetta(path='./cache')

new_cache = cache.copy(
    read=False,
    duration=timedelta(days=2),
)

Method Decorators

Use skip_self=True when decorating instance methods to exclude self from cache key hashing:

class DataService:
    @Cachetta(path='./cache.json', skip_self=True)
    def get_data(self, user_id):
        return fetch_user(user_id)

Pickle Security

Cachetta uses a restricted unpickler that only deserializes known-safe types. Raw pickle.load() allows arbitrary code execution from tampered cache files – the restricted unpickler blocks this by default.

Default safe types

Primitives (int, float, str, bytes, bool, None, list, dict, tuple), set, frozenset, complex, bytearray, range, slice, datetime (datetime, date, time, timedelta, timezone), Decimal, UUID, OrderedDict, defaultdict, deque, and pathlib paths.

Extending the allowlist

If you cache custom types (dataclasses, named tuples, etc.), add them to allowed_pickle_types:

from cachetta import Cachetta

@dataclass
class UserProfile:
    name: str
    score: float

cache = Cachetta(
    path='./cache.dat',
    allowed_pickle_types={UserProfile},
)

Custom types are merged with the defaults – you don’t lose the built-in safe types.

Error behavior

When a cache file contains a blocked type, read_cache logs a warning and yields None, the same as for corrupt data. The UnsafePickleError exception is available if you need to catch it explicitly:

from cachetta import UnsafePickleError

Error Handling

Cachetta gracefully handles corrupt cache files by yielding None:

cache = Cachetta(path='./cache.json')

with read_cache(cache) as data:
    if data is None:
        data = fetch_fresh_data()
        write_cache(cache, data)

Logging

import logging

# Enable debug logging
logging.getLogger("cachetta").setLevel(logging.DEBUG)

Configuration Reference

Option Type Default Description
path str \| Callable required Cache file path or path function
read bool True Allow reading from cache
write bool True Allow writing to cache
duration timedelta 7 days Cache TTL
lru_size int None Max in-memory LRU entries
condition Callable None Predicate to decide whether to cache
stale_duration timedelta None Time past expiry to serve stale data
skip_self bool False Exclude self from cache key hashing
allowed_pickle_types set[type] None Additional types to allow during deserialization