Caching Patterns

Overview

Multi-backend caching behind CacheBackendProtocol (Result-based: get/set/delete) plus CacheService (ergonomic facade with stampede protection). Swap memory, Redis, or Memcached via config without changing service code.

Basic Usage

from lexigram.contracts.infra.cache import CacheBackendProtocol

class ProductService:
    def __init__(self, cache: CacheBackendProtocol):
        self.cache = cache

    async def get_price(self, product_id: str) -> Result[float, DomainError]:
        cached = await self.cache.get(f"price:{product_id}")
        if cached.is_ok():
            return Ok(float(cached.unwrap()))

        price = await self._compute_price(product_id)
        await self.cache.set(f"price:{product_id}", str(price), ttl=300)
        return Ok(price)

Protocol methods: get, set(key, value, ttl=None), delete, delete_many, delete_pattern, exists, get_many, set_many, clear.

Stampede Protection (CacheService)

from lexigram.cache import CacheService

class PriceService:
    def __init__(self, cache: CacheService):
        self.cache = cache

    async def get_price(self, product_id: str) -> float:
        return await self.cache.get_or_compute(
            key=f"price:{product_id}",
            factory=lambda: self._compute_price(product_id),
            ttl=300,
        )

get_or_compute(key, factory, ttl, backend) deduplicates concurrent misses — one caller recomputes, others wait. Also available: remember(...) decorator and tag-based invalidate_by_tag(...).

Named Backends

Each configured backend gets its own named binding of both CacheBackendProtocol and CacheService. Inject a specific one with the Named marker:

from typing import Annotated
from lexigram.di.markers import Named
from lexigram.contracts.infra.cache import CacheBackendProtocol

class ReportingService:
    def __init__(
        self,
        fast_cache: Annotated[CacheBackendProtocol, Named("session")],
        slow_cache: Annotated[CacheBackendProtocol, Named("analytics")],
    ): ...

Config (cache: section)

cache:
  enabled: true
  backends:
    - name: default          # becomes the unnamed/default binding
      type: redis            # memory | redis | memcached
      default_ttl: 300
      key_prefix: "myapp:"   # all keys get this prefix
      redis_host: localhost
      redis_port: 6379
      redis_db: 0

Pattern Invalidation

# On CacheBackendProtocol / CacheService — returns count deleted
await self.cache.delete_pattern("user:*:profile")

Testing

from lexigram.testing.fakes import FakeCache   # in-memory fake backend
from lexigram.cache import CacheModule
CacheModule.stub()                             # module stub for integration tests

Common Mistakes

• No TTL on cached values — stale data forever
• Caching per-user data under a shared key — wrong data across users
• Calling get_or_compute on CacheBackendProtocol — it lives on CacheService
• Using default backend when multiple caches configured — disambiguate with Named("backend-name")
• No FakeCache in tests — tests hit real Redis or skip coverage