Pluggable caching for Python

pluca is a Python caching library for applications and libraries that need a consistent cache API across different storage backends. It includes file-based, SQLite, in-memory, and other cache backends that can be swapped with minimal code changes.

The name pluca stands for "pluggable cache architecture". The project is built around the idea that application code should be able to depend on one cache interface while choosing the storage backend that best fits each use case. Applications create and manage Cache instances directly.

In this document, backend means the underlying storage technology (file system, SQLite, memory, DBM, and similar services), and adapter means the Adapter class that talks to that backend.

Supported Python versions: 3.11+.

Development

Install the development dependency group in an editable environment with:

$ pip install --group=dev -e .

Why pluca

Unified cache interface for multiple backends
Built-in file, SQLite, and memory caches
Decorator support for caching function return values
No external runtime dependencies

Features

Unified cache interface - your application can just instantiate a Cache object and pass it around — client code just accesses the cache without having to know any of the back-end details, expiration logic, etc.
Easy interface - writing a pluca adapter for a new backend is very straightforward
It is fast - the library is developed with performance in mind
It works out-of-box - a file system cache is provided that can be used out-of-box
No batteries needed - pluca has no external dependencies

File backend

The pluca.file backend stores cache entries on the file system while keeping the same cache API used by the other backends.

>>> import pluca.file
>>> file_cache = pluca.Cache(pluca.file.Adapter(name='docs-file-cache'))
>>> file_cache.put('answer', 42)
>>> file_cache.get('answer')
42

This backend works well when you want a disk-backed cache for CLI tools, desktop applications, background jobs, or other programs that need cached values to remain available between runs.

SQLite backend

The pluca.sqlite3 backend stores cache entries in a SQLite database while preserving the same pluggable cache interface.

>>> import pluca.sqlite3
>>> import tempfile
>>> sqlite_tempdir = tempfile.TemporaryDirectory()
>>> sqlite_cache = pluca.Cache(pluca.sqlite3.Adapter(
...     filename=f'{sqlite_tempdir.name}/cache.db'))
>>> sqlite_cache.put('user-count', 123)
>>> sqlite_cache.get('user-count')
123

This backend is useful when you want a persistent local cache with a single-file database and atomic bulk writes via put_many().

Memory backend

The pluca.memory backend keeps cached values in process memory for fast repeated lookups during the life of the cache object.

>>> import pluca.memory
>>> memory_cache = pluca.Cache(pluca.memory.Adapter(max_entries=1000))
>>> memory_cache.put('greeting', 'hello')
>>> memory_cache.get('greeting')
'hello'

This backend is a good fit for temporary application caching, repeated function results, and other cases where in-memory speed matters more than persistence.

It also supports automatic maximum entry control so a cache can cap its size instead of growing until it fills all available memory.

Multiprocessing backend

The pluca.multiprocessing backend stores entries in a multiprocessing.Manager().dict() shared dictionary, so multiple processes on the same host can use one in-memory cache.

This backend is useful for local multi-process workloads that need shared ephemeral cache data without writing to disk.

Because operations go through manager IPC/proxy calls, this backend is typically slower than the in-process memory backend for high-throughput single-process access.

The full list of built-in adapters is available in the Included adapters section below.

Use cases

Add Python caching to an application without coupling code to one storage engine
Use a file-backed cache for local persistent caching
Use SQLite for persistent cache storage in a single database file
Use an in-memory cache for fast in-process lookups
Use a manager-backed cache to share in-memory values across local worker processes
Swap cache backends without changing application cache logic
Cache expensive calculations or function return values

How to use

First import the cache module:

>>> import pluca.file  # Use a file system cache.

Now create the cache object:

>>> cache = pluca.Cache(pluca.file.Adapter())

Store 3.1415 in the cache using pi as key:

>>> cache.put('pi', 3.1415)

Now retrieve the value from the cache.

>>> pi = cache.get('pi')
>>> pi
3.1415
>>> type(pi)
<class 'float'>

Non-existent or expired cache entries raise KeyError.

>>> cache.get('notthere')
Traceback (most recent call last):
    ...
KeyError: 'notthere'

Use remove() to delete entries from the cache:

>>> cache.put('foo', 'bar')
>>> cache.get('foo')
'bar'
>>> cache.remove('foo')
>>> cache.get('foo')
Traceback (most recent call last):
    ...
KeyError: 'foo'

To test if an entry exists, use has():

>>> cache.put('this', 'is in the cache')
>>> cache.has('this')
True
>>> cache.has('that')
False

You can provide a default value for when the key does not exist or has expired. The method will not raise KeyError in this case, it will return the default value instead.

>>> cache.get('notthere', 12345)
12345

By default cache entries are set to “never” expire — cache adapters can expire entries though, for example to use less resource. Here’s an example of how to store a cache entry with an explicit expiration time:

>>> cache.put('see-you', 'in two secs', 1)  # Expire in 1 second.
>>> import time; time.sleep(1)  # Wait for it to expire.
>>> cache.get('see-you')
Traceback (most recent call last):
    ...
KeyError: 'see-you'

Passing max_age=0 marks an entry as immediately expired, while max_age=None keeps the default behavior (no explicit expiration).

Cache keys can be any object (but see Caveats below):

>>> key = (__name__, True, 'this', 'key', 'has', 'more', 'than', 1, 'value')
>>> cache.put(key, 'data')
>>> cache.get(key)
'data'

Cached values can be any pickable data:

>>> import datetime
>>> alongtimeago = datetime.date(2020, 1, 1)
>>> cache.put('alongtimeago', alongtimeago)
>>> today = cache.get('alongtimeago')
>>> today
datetime.date(2020, 1, 1)
>>> type(today)
<class 'datetime.date'>

Flushing the cache removes all entries:

>>> cache.put('bye', 'tchau')
>>> cache.flush()
>>> cache.get('bye')
Traceback (most recent call last):
    ...
KeyError: 'bye'

Calling flush() on a fresh cache with no stored entries is safe and acts as a no-op.

Abstracting cache adapters

Here’s how to abstract cache adapters. First, let’s define a function that calculates a factorial. The function also receives a cache object to store results, so that the calculation results are cached.

>>> from math import factorial
>>> def cached_factorial(cache, n):
...     try:
...         res = cache.get(('factorial', n))
...     except KeyError:
...         print(f'CACHE MISS - calculating {n}!')
...         res = factorial(n)
...         cache.put(('factorial', n), res)
...     return res

Now let’s try this with the file cache created above. First call should be a cache miss:

>>> cached_factorial(cache, 10)
CACHE MISS - calculating 10!
3628800

Subsequent calls should get the results from the cache:

>>> cached_factorial(cache, 10)
3628800

Now let's switch to the null adapter (it does not store data anywhere - see help(pluca.null.Adapter) for more info):

>>> import pluca.null
>>> null_cache = pluca.Cache(pluca.null.Adapter())
>>>
>>> cached_factorial(null_cache, 10)
CACHE MISS - calculating 10!
3628800

Using caches as decorators

Caches can also be used as decorator to cache function return values:

>>> @cache
... def expensive_calculation(alpha, beta):
...     res = 0
...     print('Doing expensive calculation')
...     for i in range(0, alpha):
...         for j in range(0, beta):
...             res = i * j
...     return res
>>>
>>> cache.flush()  # Let's start with an empty cache.
>>>
>>> expensive_calculation(10, 20)
Doing expensive calculation
171

Calling the function again with the same parameters returns the cached result:

>>> expensive_calculation(10, 20)
171

Each function can have their own expiration:

>>> @cache(max_age=1)  # Expire after one second.
... def quick_calculation(alpha, beta):
...     print(f'Calculating {alpha} + {beta}')
...     return alpha + beta

First call executes the function. Second call gets the cached value.

>>> quick_calculation(1, 2)
Calculating 1 + 2
3
>>> quick_calculation(1, 2)
3

After the expiry time the calculation is done again:

>>> import time; time.sleep(1)
>>> quick_calculation(1, 2)
Calculating 1 + 2
3

Miscellaneous cache methods

`get_put()`

Use get_put() to conveniently get a value from the cache, or call a function to generate it, if it is not cached already:

>>> cache.flush()
>>>
>>> def calculate_foo():
...    print('Calculating foo')
...    return 'bar'
>>>
>>> cache.get_put('foo', calculate_foo)
Calculating foo
'bar'

>>> cache.get_put('foo', calculate_foo)
'bar'

get_put() also supports dependency-based invalidation via dependency=. A dependency is a callable returning a value that represents external state. When that value changes, the cached entry is recomputed:

>>> import pluca.memory
>>> import pluca.invalidation
>>> dep_cache = pluca.Cache(pluca.memory.Adapter())
>>> source = {'version': 1}
>>> calls = [0]
>>> def render():
...     calls[0] += 1
...     return f'value-{calls[0]}'
>>> dep_cache.get_put('foo', render,
...                   dependency=lambda: source['version'])
'value-1'
>>> dep_cache.get_put('foo', render,
...                   dependency=lambda: source['version'])
'value-1'
>>> source['version'] = 2
>>> dep_cache.get_put('foo', render,
...                   dependency=lambda: source['version'])
'value-2'

The helper module pluca.invalidation provides stdlib-only probes for common dependency sources, such as file mtime (file_mtime()), SQLite scalar queries (sqlite_scalar()), environment variables (env_var()), and probe composition (combine()) when you need to track multiple dependency sources as one value.

combine() accepts an operator= argument backed by the CombineOperator enum. The default is CombineOperator.OR (invalidate when any probe changes). Use CombineOperator.AND to invalidate only when all combined probes change relative to the cached dependency snapshot.

Dependency support is enabled by default on every cache object via enable_dependencies=True. To skip dependency checks for performance, initialize the cache with enable_dependencies=False:

>>> dep_disabled = pluca.Cache(pluca.memory.Adapter(),
...                            enable_dependencies=False)

When disabled, calls that request dependency= raise pluca.CacheConfigurationError.

`set_max_age()`

Use set_max_age() to update the expiration of an existing key without recomputing or replacing its value:

>>> cache.put('session', {'user': 'alice'}, max_age=1)
>>> cache.set_max_age('session', max_age=60)

Working with multiple entries

You can add many entries to the cache at once by calling put_many():

>>> cache.put_many({'foo': 'bar', 'zee': 'too'})
>>> cache.get('zee')
'too'

You can also pass an iterable of (key, value) tuples. This is useful for caching with non-hashable keys:

>>> cache.put_many([(['a', 'b', 'c'], 123), ('pi', 3.1415)])
>>> cache.get(['a', 'b', 'c'])
123

On the sqlite3 backend, put_many() is atomic: all rows are written in a single transaction and committed once. If one row fails, no rows from that put_many() call are persisted.

New sqlite3 cache tables are created with SQLite WITHOUT ROWID.

Use get_many() to get many results at once. This method returns a list of (key, value) tuples:

>>> cache.get_many(['zee', 'pi'])
[('zee', 'too'), ('pi', 3.1415)]

get_many() returns a list of tuples (instead of a dict) so it can support keys that are not hashable. This makes it safe for cases like list or dict keys, where building a dict would fail.

When all returned keys are hashable and unique, you can convert the result to a dict:

>>> dict(cache.get_many(['zee', 'pi']))
{'zee': 'too', 'pi': 3.1415}

Keep in mind that dict conversion requires hashable keys and will collapse duplicate keys to the last value.

Notice that get_many() does not raise KeyError when a key is not found or has expired. Instead, the key will not be present in the returned list:

>>> cache.get_many(['pi', 'not-there'])
[('pi', 3.1415)]

However, you can pass a default value to get_many(). This value will be returned for any non-existing keys:

>>> cache.get_many(['pi', 'not-there', 'also-not-there'], default='yes')
[('pi', 3.1415), ('not-there', 'yes'), ('also-not-there', 'yes')]

Use remove_many() to remove multiple keys at once. Missing keys are ignored:

>>> cache.put_many({'x': 1, 'y': 2})
>>> cache.remove_many(['x', 'not-there'])
>>> cache.get('x')
Traceback (most recent call last):
    ...
KeyError: 'x'
>>> cache.get('y')
2

Garbage collection

Garbage collection tells the cache to remove expired entries to save resources. This is done by the gc() method:

>>> cache.gc()

Notice that pluca never calls gc() automatically — it is up to your application to call it eventually to do garbage collection.

Calling gc() on a fresh cache is also safe and behaves as a no-op.

Composite caches

The pluca.comp adapter chains multiple caches into a single cache. Writes go to every configured child cache, reads return the first hit, and remove() attempts deletion on every configured child cache, raising KeyError only when the key is missing from all tiers.

>>> import pluca.comp
>>> import pluca.memory
>>> import pluca.file
>>> comp_cache = pluca.Cache(pluca.comp.Adapter())
>>> comp_cache.add_cache(pluca.Cache(pluca.memory.Adapter(max_entries=100)))
>>> comp_cache.add_cache(pluca.Cache(pluca.file.Adapter(name='comp-example')))

Composite children are created explicitly and added in tier order:

>>> cfg_adapter = pluca.comp.Adapter()
>>> cfg_adapter.add_cache(pluca.Cache(
...     pluca.memory.Adapter(max_entries=10)))
>>> cfg_adapter.add_cache(pluca.Cache(pluca.null.Adapter()))
>>> cfg_cache = pluca.Cache(cfg_adapter)

Concurrency

A cache object created in a thread and used only by that same thread is a safe usage pattern.

Backend notes

Guidelines by backend:

pluca.memory: safe for thread-confined instances; do not share one instance across threads without external locking.
pluca.sqlite3: uses one SQLite connection per adapter. With default SQLite settings, that connection is thread-affine, so use it only in the thread that created it unless you explicitly configure otherwise and serialize access yourself.
pluca.dbm: no internal locking for shared concurrent access.
pluca.file: supports file-level locking (locking=), which helps coordinate entry file access, but does not make all cache-level operations atomic across threads.
pluca.multiprocessing: designed for cross-process sharing through a manager-backed dictionary.

`get_put()` and dependencies

get_put() is not single-flight: concurrent misses may call the producer more than once and race to store the final value.

The same caveat applies when using dependency= in get_put() or on the cache decorator: dependency state is checked and cached in separate steps, so concurrent calls can still recompute and overwrite values.

If at-most-once computation per key matters, use your own lock around the entire get_put() call (or around the decorated function call path), usually with a key-scoped threading.Lock.

Caveats

Cache keys are internally mapped using the repr() of (type(key), key), then hashed. As long as your key objects have stable representations, this will cause no problems. However, for types with unstable representation, for example those that have no inherent ordering (e.g., frozenset), this can be problematic because there’s no guarantee that repr((type(key), key)) will return the same string value every time. This applies even to objects deep inside your key. The type is part of the mapping, so 1 and '1' are distinct keys. For example, this is a bad composite key:
```
  >>> key = ('foo', ('another', set((1, 2, 3))))  # set is unstable
```
By default pluca uses pickle to serialize and unserialize data. A quote from the Python documentation:

It is possible to construct malicious pickle data which will execute arbitrary code during unpickling. Never unpickle data that could have come from an untrusted source, or that could have been tampered with.

So be careful where you store your cached data.
The sqlite3 backend only accepts simple SQL identifiers for dynamic names used in statements (for example PRAGMA names). Identifiers must match [A-Za-z_][A-Za-z0-9_]*; invalid names raise ValueError during cache initialization.
The file backend defaults to name='pluca'. If cache_dir is not provided, it uses appdirs.user_cache_dir() when appdirs is installed, otherwise ~/.cache. The cache name must be a single safe path segment: it cannot be absolute, cannot contain / or \\, and cannot be . or ...

File locking can be controlled with the locking argument:
- locking='auto' (default) selects the most efficient stdlib lock mechanism for the current OS.
- locking=None disables file locking.
- locking='mkdir' uses lock directories and is suitable when cache files are on NFS.
- locking='flock' (POSIX) and locking='msvcrt' (Windows) force a specific stdlib lock mechanism.
For locking='mkdir', these options control lock waiting and stale lock cleanup:
- mkdir_stale_age (default 300.0 seconds)
- mkdir_wait_timeout (default 30.0 seconds)
- mkdir_poll_interval (default 0.05 seconds)
Lock ownership metadata is written to <entry>.lock/owner as three newline-separated values: PID, hostname, and creation timestamp.

Locks are applied to each entry file. On POSIX (flock), reads use a shared lock and writes/removals use an exclusive lock. On Windows (msvcrt), reads and writes both use exclusive locking.

pluca.utils.create_cachedir_tag() can create a CACHEDIR.TAG file for cache directories managed by your application:

  >>> import tempfile
  >>> import pluca.utils
  >>> tmp = tempfile.TemporaryDirectory()
  >>> pluca.utils.create_cachedir_tag(tmp.name)

Included adapters

These are the cache adapters that come with the pluca package:

file adapter - stores cache entries on the file system backend.
sqlite3 adapter - stores cache entries in a SQLite3 backend.
memory adapter - stores cache entries in process memory.
multiprocessing adapter - stores cache entries in a manager-backed shared dictionary for cross-process local access.
comp adapter - composes multiple caches into a tiered cache.
dbm adapter - stores cache entries using a DBM backend.
null adapter - never persists values and get() always raises KeyError.

The core package supports SQLite for SQL storage.

To obtain help about those cache adapters, run help(pluca.MODULE.Adapter), where MODULE is one of the module names above.

Benchmarking

The pluca.benchmark module can be used to benchmark the adapters:

$ python -m pluca.benchmark

Pass -h to see the benchmark options.

For deterministic stdlib-only behavior across platforms, DBM benchmarking uses dbm.dumb.

Issues? Bugs? Suggestions?

Visit: https://github.com/flaviovs/pluca

Name		Name	Last commit message	Last commit date
.agents/skills		.agents/skills
src/pluca		src/pluca
tests		tests
.gitignore		.gitignore
ADAPTER_SPEC.md		ADAPTER_SPEC.md
AGENTS.md		AGENTS.md
CHANGELOG.md		CHANGELOG.md
LICENSE.txt		LICENSE.txt
README.md		README.md
pyproject.toml		pyproject.toml

Navigation Menu

Repository files navigation

Pluggable caching for Python

Development

Why pluca

Features

File backend

SQLite backend

Memory backend

Multiprocessing backend

Use cases

How to use

Abstracting cache adapters

Using caches as decorators

Miscellaneous cache methods

`get_put()`

`set_max_age()`

Working with multiple entries

Garbage collection

Composite caches

Concurrency

Backend notes

`get_put()` and dependencies

Caveats

Included adapters

Benchmarking

Issues? Bugs? Suggestions?

About

Used by

Contributors

Languages

Latest commit

History

Folders and files

Repository files navigation

Pluggable caching for Python

Development

Why pluca

Features

File backend

SQLite backend

Memory backend

Multiprocessing backend

Use cases

How to use

Abstracting cache adapters

Using caches as decorators

Miscellaneous cache methods

get_put()

set_max_age()

Working with multiple entries

Garbage collection

Composite caches

Concurrency

Backend notes

get_put() and dependencies

Caveats

Included adapters

Benchmarking

Issues? Bugs? Suggestions?

About

Topics

Resources

Stars

Watchers

Forks

Used by

Contributors

Languages

`get_put()`

`set_max_age()`

`get_put()` and dependencies