caching

caching

Covers Redis as cache, pub/sub bus, and job queue — plus lightweight in-process caching for Node/Python services.

1) Redis connection (Python — async)

import redis.asyncio as aioredis

redis = aioredis.from_url(
    "redis://localhost:6379",
    encoding="utf-8",
    decode_responses=True,
    max_connections=20,
)

# Simple get/set with TTL
await redis.set("key", "value", ex=300)   # expires in 5 min
value = await redis.get("key")            # None if expired/missing

# Delete
await redis.delete("key")

2) Cache-aside pattern (Python)

import json

async def get_article(slug: str) -> dict:
    cache_key = f"article:{slug}"

    cached = await redis.get(cache_key)
    if cached:
        return json.loads(cached)

    article = await db.fetch_article(slug)        # expensive DB read
    await redis.set(cache_key, json.dumps(article), ex=3600)
    return article

# Invalidate on write
async def update_article(slug: str, data: dict):
    await db.update_article(slug, data)
    await redis.delete(f"article:{slug}")         # bust cache

3) Pub/sub (event bus between agents)

# Publisher
async def publish(channel: str, payload: dict):
    await redis.publish(channel, json.dumps(payload))

await publish("article.ready", {"slug": slug, "domain": domain})

# Subscriber (runs in background task)
async def subscribe(channel: str):
    pubsub = redis.pubsub()
    await pubsub.subscribe(channel)
    async for message in pubsub.listen():
        if message["type"] == "message":
            data = json.loads(message["data"])
            await handle_event(data)

4) Redis as job queue (simple LPUSH/BRPOP)

QUEUE = "jobs:scrape"

# Enqueue
await redis.lpush(QUEUE, json.dumps({"url": url, "domain": domain}))

# Worker — blocking pop, 30s timeout
async def worker():
    while True:
        item = await redis.brpop(QUEUE, timeout=30)
        if item:
            _, raw = item
            job = json.loads(raw)
            await process(job)