ZerdaZerda

cache.storage.base_storage

On this page

BaseStorage

Python
class BaseStorage(ABC)

Overview

BaseStorage is the abstract contract that every persistence backend must implement. It exists so that CacheManager can be written once against a uniform interface while build_storage() swaps in MemoryStorage, SQLiteStorage, or RedisStorage based on CacheConfig.storage_backend.

All methods operate on IntelligentCacheEntry — BaseStorage is specific to the semantic-cache pipeline and is not used by MultiLevelCache, which manages its own storage internally.

Abstract methods

SignaturePurpose
get(key: str) -> Optional[IntelligentCacheEntry]Fetch an entry by its exact cache key, or None if absent/expired
set(key: str, entry: IntelligentCacheEntry, ttl_s: Optional[float] = None) -> NoneInsert or overwrite an entry, optionally with a TTL
delete(key: str) -> boolRemove an entry; returns whether it existed
exists(key: str) -> boolCheck presence without deserializing the full entry (implementations vary in how cheap this actually is)
keys_by_tenant(tenant_id: str) -> List[str]List all keys belonging to a tenant
all_keys() -> List[str]List all non-expired keys
total_size_bytes() -> intApproximate total storage footprint
flush_tenant(tenant_id: str) -> intDelete every entry for a tenant; returns count removed
close() -> NoneRelease any held resources (connections, pools)

Every subclass implements all nine methods; there are no optional/default implementations on the ABC itself (each is declared with ... as its body).

Implementations

ClassBackendPersistenceNotes
MemoryStoragePython dictNone — lost on process exitSimplest, fastest, no I/O
SQLiteStoragesqlite3 (WAL mode)Single file on diskDefault backend (StorageBackend.SQLITE)
RedisStorageredis clientExternal Redis serverFalls back to SQLiteStorage if Redis is unreachable

Extending: implementing a custom backend

Because BaseStorage is a plain abc.ABC with @abstractmethod, any class that implements all nine methods can be used wherever a BaseStorage is expected — including by constructing a CacheManager and manually replacing its internal _storage attribute (there is no public constructor parameter to inject a custom storage instance directly; see Known Issues for the implication of this).

Python · 37 lines
from zerda.cache.storage import BaseStorage
from zerda.cache.models import IntelligentCacheEntry
from typing import List, Optional

class MyCustomStorage(BaseStorage):
    def __init__(self):
        self._data = {}

    def get(self, key: str) -> Optional[IntelligentCacheEntry]:
        return self._data.get(key)

    def set(self, key, entry, ttl_s=None) -> None:
        self._data[key] = entry

    def delete(self, key: str) -> bool:
        return self._data.pop(key, None) is not None

    def exists(self, key: str) -> bool:
        return key in self._data

    def keys_by_tenant(self, tenant_id: str) -> List[str]:
        return [k for k, e in self._data.items() if e.tenant_id == tenant_id]

    def all_keys(self) -> List[str]:
        return list(self._data.keys())

    def total_size_bytes(self) -> int:
        return sum(e.size_bytes for e in self._data.values())

    def flush_tenant(self, tenant_id: str) -> int:
        keys = self.keys_by_tenant(tenant_id)
        for k in keys:
            self.delete(k)
        return len(keys)

    def close(self) -> None:
        pass

Related pages