Guide

Requirements

Package	Required	Purpose
lexigram	Yes	Core framework
lexigram-contracts	Yes	Protocol definitions
redis	Recommended	Redis cache backend
pymemcache	Optional	Memcached cache backend

Overview

lexigram-cache provides a unified caching API across multiple backends. It handles the common caching patterns so you don’t have to — stampede protection, TTL management, serialization, tag-based invalidation, and health checks.

Mental Model

Application Code
     │
     ▼
  CacheService  ←  unified API (get/set/delete/delete_pattern)
     │
     ├── StampedeProtectedCache  ←  lock-based stampede prevention
     │
     ▼
  CacheBackendProtocol  ←  backend abstraction (Result-based)
     │
     ├── MemoryCacheBackend  (in-process, no deps)
     ├── RedisCacheBackend   (requires redis-py)
     └── MemcachedCacheBackend  (requires pymemcache)

Core Concepts

Backend Abstraction

All backends implement CacheBackendProtocol from lexigram.contracts.infra.cache. The protocol returns Result[T, CacheError] for every operation:

from lexigram.contracts.infra.cache import CacheBackendProtocol
from lexigram.result import Ok, Err

# Backend returns Result — CacheService unwraps it internally
result = await backend.get("my-key")
if result.is_ok():
    value = result.unwrap()

Backend	Extra	Storage
MemoryCacheBackend	None	In-process dict
RedisCacheBackend	lexigram-cache[redis]	Redis server
MemcachedCacheBackend	lexigram-cache[memcached]	Memcached server

CacheService (High-Level API)

CacheService wraps the backend and provides ergonomic access:

from lexigram.cache import CacheService

# Resolved from the container — inject via constructor
await cache.set("user:42", {"name": "Alice"}, ttl=300)
value = await cache.get("user:42")        # → {"name": "Alice"}
await cache.delete("user:42")             # → True
count = await cache.delete_pattern("user:*")  # → number of keys

CacheService.get() returns the raw value or default (not Result). Errors are logged and return the default — the service prefers graceful degradation over crashing.

Named Backends

Configure multiple backends with different names in CacheConfig.backends. The first default: true backend is the default:

cache:
  backends:
    - name: "hot"
      type: memory
      default: true
    - name: "persistent"
      type: redis
      host: localhost
      port: 6379

Resolve a specific backend:

hot_cache = await container.resolve(CacheService, name="hot")
persistent_cache = await container.resolve(CacheService, name="persistent")

Stampede Protection

When service.enable_protection is True (default), CacheService uses lock-based stampede protection. Only one process recomputes the value while others wait:

cache.service:
  enable_protection: true
  protection_lock_ttl: 30
  protection_max_wait: 10.0

Tag-Based Invalidation

Tag cache entries so you can invalidate groups of keys:

from lexigram.cache import CacheService

await cache.set("article:1", data, tags=["articles", "breaking"])
await cache.set("article:2", data, tags=["articles"])

# Invalidate all articles
await cache.invalidate_tags(["articles"])

# The next get() for article:1 returns None

Serialization

By default, CacheService serializes complex objects to JSON. For backward compatibility, pickle is available but must be explicitly opted in:

cache.service:
  default_serializer: "json"    # default
  allow_pickle: false           # opt-in required for pickle

Available serializers: JSONSerializer, PickleSerializer, CompressingSerializer.

Decorator Syntax

Decorate async functions with @cache or @cacheable:

from lexigram.cache import cache, cacheable


@cache(ttl=300, tags=["user"])
async def get_user(user_id: str) -> dict:
    return await db.fetch_user(user_id)


@cacheable(ttl=60)
async def expensive_computation(input: str) -> str:
    # result is cached automatically
    return await compute(input)

Typical Usage

1. Wire the cache

from lexigram import Application
from lexigram.cache import CacheModule
from lexigram.cache.config import CacheConfig


def create_app() -> Application:
    app = Application(name="my-app")
    app.add_module(CacheModule.configure(CacheConfig(
        backends=[{
            "name": "default",
            "type": "memory",
            "default": True,
        }],
    )))
    return app

2. Use in a service

from lexigram.di import inject
from lexigram.cache import CacheService


class UserService:
    @inject
    def __init__(self, cache: CacheService) -> None:
        self.cache = cache

    async def get_user(self, user_id: str) -> dict:
        cached = await self.cache.get(f"user:{user_id}")
        if cached is not None:
            return cached
        user = await self._load_from_db(user_id)
        await self.cache.set(f"user:{user_id}", user, ttl=300)
        return user

Best Practices

• ✅ Use MemoryCacheBackend for testing — no external dependencies
• ✅ Set default_ttl on backends to prevent unbounded cache growth
• ✅ Use tag-based invalidation for group cache clearing
• ✅ Enable stampede protection for expensive-to-compute values
• ⚠️ Opt in to pickle only if you need to cache non-JSON-serializable objects
• ❌ Don’t cache user secrets (passwords, tokens) in plain text
• ❌ Don’t skip TTL for volatile data — always set an expiration

Next Steps

• How-Tos — Redis setup, stampede protection, multi-backend
• Configuration — all configuration keys
• Troubleshooting — common errors and fixes