BaseCache & Caching Layer
API reference for BaseCache with Redis integration, Pydantic deserialization, tunable defaults, and in-memory TTLLRUCache fallback.
BaseCache provides a unified caching interface supporting distributed caching (via Redis) with an asynchronous, thread-safe local memory fallback (via TTLLRUCache). It coordinates global cache lifecycles, structured serialization, and periodic background eviction routines safely within active event loops.
Class Definition
from typing import Generic, TypeVar
from zcore.cache import BaseCache
T = TypeVar("T")
class BaseCache(Generic[T]):
def __init__(
self,
prefix: str,
local_maxsize: int | None = None,
default_ttl: int | None = None
) -> None:
...Initialization Parameters
Prop
Type
Properties
Prop
Type
Methods
get
Retrieves and deserializes a cache record by its key. Looks up records from Redis first, falling back to the local memory store on connection errors or missing pools.
async def get(
self,
key: str,
target_type: type[BaseModel] | None = None
) -> T | BaseModel | Any | None: ...Prop
Type
Automatic Pydantic Validation: If target_type is provided (e.g., TaskResponse), BaseCache automatically deserializes the raw JSON and invokes target_type.model_validate(parsed_data).
set
Serializes and persists a key-value record with a Time-To-Live (TTL). Stores in Redis if available, falling back to the thread-safe local memory buffer on network failures.
async def set(self, key: str, value: T, ttl: int | None = None) -> None: ...Prop
Type
delete
Evicts a record from both distributed Redis and local memory fallback caches.
async def delete(self, key: str) -> None: ...Prop
Type
Global Cache Functions
init_cache
Configures the shared distributed Redis client and attempts to schedule the background in-memory garbage collection worker loop if an active event loop is running (with lazy scheduling ensuring execution across subsequent cache operations).
from zcore.cache import init_cache
def init_cache(redis_url: str | None = None, **kwargs: Any) -> None: ...Prop
Type
close_cache
Cancels background eviction worker loops and cleanly closes shared distributed Redis connections.
from zcore.cache import close_cache
async def close_cache() -> None: ...Local In-Memory Fallback: TTLLRUCache
Thread-safe in-memory cache featuring granular record-level expiration boundaries and LRU eviction, dynamically bounded by settings:
from zcore.cache import TTLLRUCache
class TTLLRUCache:
def __init__(self, maxsize: int | None = None) -> None: ...
def get(self, key: str) -> Any | None: ...
def set(self, key: str, value: Any, ttl: int | None = None) -> None: ...
def delete(self, key: str) -> None: ...
def evict_expired(self) -> None: ...
@classmethod
def evict_all_expired(cls) -> None: ...Prop
Type