How to cache database queries

Accelerate queries using BaseCache with automatic Pydantic deserialization, tunable settings, and resilient local LRU fallback.

ZCore's BaseCache provides a unified caching interface. It connects to distributed Redis instances while seamlessly degrading to a thread-safe in-memory TTLLRUCache if Redis is offline or unconfigured.

1. Initialize the Cache Client

Configure the global cache client in your main.py (or within a plugin lifecycle hook):

# main.py
from zcore import settings
from zcore.cache import init_cache

# Initializes distributed Redis connection and starts background local eviction loop
init_cache(redis_url=settings.REDIS_URL)

2. Implement the Cache-Aside Pattern

Instantiate BaseCache with a domain prefix namespace. BaseCache automatically derives default TTLs and local size boundaries from your Settings, and accepts target_type to auto-deserialize JSON into validated Pydantic models:

# services.py or repositories.py
import uuid
from zcore.cache import BaseCache
from .schemas import TaskResponse
from .models import Task

class TaskService:
    def __init__(self, repo):
        self.repo = repo
        # Inherits default TTL from settings.CACHE_DEFAULT_TTL (default: 3600s)
        # and local maxsize from settings.CACHE_LOCAL_MAXSIZE (default: 1000 items)
        self.cache = BaseCache[TaskResponse](prefix="tasks")

    async def get_task_details(self, task_id: uuid.UUID) -> TaskResponse:
        cache_key = str(task_id)

        # 1. Attempt to fetch from Cache (Auto-deserializes into TaskResponse)
        cached_data = await self.cache.get(cache_key, target_type=TaskResponse)
        if cached_data:
            return cached_data

        # 2. Cache Miss: Fetch from database
        db_record = await self.repo.get(id=task_id)
        task_response = TaskResponse.model_validate(db_record)

        # 3. Populate Cache (ttl is optional; falls back to instance default_ttl)
        await self.cache.set(cache_key, task_response)
        
        return task_response

3. Invalidate Cache on Mutations

Purge cached records when modifying or deleting database entities:

async def update_task(self, task_id: uuid.UUID, schema):
    updated = await self.repo.update(task_id, schema)
    
    # Evict key from both Redis and local memory fallback store
    await self.cache.delete(str(task_id))
    return updated

Dynamic Framework Settings

ZCore manages caching boundaries centrally via environment variables and the Settings class:

Setting Key	Default	Description
`CACHE_DEFAULT_TTL`	`3600` (1 hour)	Default fallback TTL in seconds when `ttl` is omitted in `cache.set()`.
`CACHE_LOCAL_MAXSIZE`	`1000`	Maximum entries retained simultaneously in the local in-memory `TTLLRUCache`.
`CACHE_EVICTION_INTERVAL`	`60`	Duration in seconds between automated garbage collection sweeps on expired keys.

Fail-Safe In-Memory Fallback: If Redis is unavailable or unconfigured (REDIS_URL=None), BaseCache automatically stores records in an in-memory TTLLRUCache. Expired keys are periodically swept by a non-blocking background worker loop scheduled at settings.CACHE_EVICTION_INTERVAL seconds.

1. Initialize the Cache Client

2. Implement the Cache-Aside Pattern

3. Invalidate Cache on Mutations

Dynamic Framework Settings

On this page