Skip to main content

Documentation Index

Fetch the complete documentation index at: https://mintlify.com/ismael-sarmiento/kimera_python/llms.txt

Use this file to discover all available pages before exploring further.

Local caching in Kimera Core stores key-value pairs entirely within the running Python process — no external service required. Two concrete engines are available: InMemoryCacheEngine, which stores data in a plain dictionary on the instance, and ThreadCacheEngine, which attaches the cache dictionary to the currently running thread. Use InMemoryCacheEngine for process-wide shared state (with a singleton pattern) and ThreadCacheEngine when each request or worker thread must maintain completely isolated cache namespaces.

​
Class Hierarchy

All cache engines descend from a two-level abstract hierarchy:

BaseCacheEngine (ABC)

Top-level abstract class. Declares the five core methods every engine must implement: get, set, delete, exists, and clear.

LocalCacheEngine (ABC)

Extends BaseCacheEngine. Adds size_regulator, normalize_timeout, and normalize_size — helpers used by both local engine implementations.
InMemoryCacheEngine and ThreadCacheEngine both extend LocalCacheEngine and provide concrete storage backends.

​
Default Values

Both local engines share the same class-level defaults defined on BaseCacheEngine:
ConstantValueMeaning
DEFAULT_TIMEOUT300.0Seconds a key lives before expiry
DEFAULT_SIZE300Maximum number of entries in the cache
These values are applied automatically when timeout or size are omitted or None in a set call.

​
InMemoryCacheEngine

InMemoryCacheEngine stores key-value pairs in a plain dict (self.memory_cache) that is owned by the engine instance. The dictionary is created in __init__ and lives for the lifetime of that instance.
Because InMemoryCacheEngine keeps state on the instance, two separate InMemoryCacheEngine() objects do not share cached data. To share cache across multiple call sites in the same process, either pass a single instance around or wrap the class in a Singleton pattern.

​
Methods

Returns the cached value if key exists and its TTL has not expired. If the key is missing or expired, delete is called and None is returned.
Stores or updates {key: value}. Before writing, size_regulator is called to evict the oldest entries if the cache is over capacity. timeout defaults to DEFAULT_TIMEOUT (300 s) and size defaults to DEFAULT_SIZE (300) when None is passed.
Returns True if key is present in the cache and its stored timeout is greater than the current epoch time. Returns False otherwise.
Removes key from the cache dictionary if it currently exists (checked via exists).
Replaces self.memory_cache with a fresh empty dict, wiping all entries.

​
Example

from components.storage.dynamic.cache.local.engines import InMemoryCacheEngine

cache = InMemoryCacheEngine()

# Store with a 10-second TTL, max 100 entries
cache.set('session:abc', {'user': 'alice'}, timeout=10, size=100)

if cache.exists('session:abc'):
    data = cache.get('session:abc')
    print(data)  # {'user': 'alice'}

cache.delete('session:abc')
cache.clear()

​
ThreadCacheEngine

ThreadCacheEngine attaches the cache dictionary to threading.currentThread() as the attribute thread_local_cache. Every thread in the process gets its own isolated dictionary — keys written in one thread are invisible to all other threads. This makes ThreadCacheEngine a natural fit for per-request caching in multi-threaded web servers.
Because the cache lives on the thread object, you can safely create a new ThreadCacheEngine() instance anywhere in the same thread and it will always read from the same underlying dictionary.

​
Methods

The API surface of ThreadCacheEngine is identical to InMemoryCacheEngine:
Retrieves the value from the current thread’s cache dictionary. Returns None (and calls delete) if the key is missing or expired.
Writes {key: value} to the current thread’s cache after running size_regulator.
Returns True if the key is present in the current thread’s cache and has not expired.
Removes the key from the current thread’s cache if it exists.
Resets the current thread’s thread_local_cache attribute to an empty dict.

​
Example

from components.storage.dynamic.cache.local.engines import ThreadCacheEngine

cache = ThreadCacheEngine()
cache.set('request_id', '1234', timeout=30)
print(cache.get('request_id'))  # '1234'

​
Cache Decorators

Kimera Core ships two function decorators that automatically store a function’s return value in a local cache engine after every call.
Wraps a function and writes its return value into the current thread’s ThreadCacheEngine under the given key.
@thread_local_cache(key, timeout=None, size=None)
def my_function(...):
    ...

​
Decorator Example

from components.storage.dynamic.cache.local.engines import thread_local_cache, ThreadCacheEngine

@thread_local_cache('computed_value', timeout=60)
def expensive_computation(x):
    return x * 42

result = expensive_computation(10)
print(ThreadCacheEngine().get('computed_value'))  # 420
@in_memory_local_cache creates a new InMemoryCacheEngine() instance each time the decorated function is called. The stored value is therefore only retrievable from that same transient instance unless you have made InMemoryCacheEngine a Singleton. For reliable cross-call retrieval, use @thread_local_cache instead, or inject a shared InMemoryCacheEngine singleton manually.

​
Size Regulation

Before every set call, size_regulator(size) is invoked. If len(cache) > size, the engine calculates how many entries exceed the limit and removes that many by calling dict.popitem() in a loop. Because Python 3.7+ dictionaries maintain insertion order, popitem() removes the most recently inserted item — effectively evicting the newest entries when the cache overflows.
popitem() removes from the end of insertion order in Python 3.7+. If you rely on LRU (least-recently-used) eviction semantics, consider using collections.OrderedDict or a dedicated LRU cache instead.

​
TTL Normalization

Raw seconds passed as timeout are converted to an absolute epoch timestamp by normalize_timeout: 
def normalize_timeout(self, timeout: float) -> float:
    timeout = float(timeout) if timeout else self.DEFAULT_TIMEOUT
    return time() + timeout
The stored value is time() + timeout. When a key is read, result['timeout'] > time() checks whether the stored absolute timestamp is still in the future. This design means no background thread is needed to expire keys — expiry is checked lazily on every get or exists call.

Build docs developers (and LLMs) love

Get started for freeTalk to us