Redis Patterns

import redisimport json
r = redis.Redis(host='localhost', port=6379, decode_responses=True)
def get_product(product_id: int):    cache_key = f"product:{product_id}"    cached = r.get(cache_key)
    if cached:        return json.loads(cached)
    product = db.query("SELECT * FROM products WHERE id = %s", product_id)    r.setex(cache_key, 3600, json.dumps(product))  # TTL: 1 hour    return product

def update_product(product_id: int, data: dict):    # DB書き込み先    db.execute("UPDATE products SET ... WHERE id = %s", product_id)
    # キャッシュを即座に更新    cache_key = f"product:{product_id}"    r.setex(cache_key, 3600, json.dumps(data))

# タグベース削除 — セット内で関連キーをグループ化def cache_product(product_id: int, category_id: int, data: dict):    key = f"product:{product_id}"    tag = f"tag:category:{category_id}"    pipe = r.pipeline(transaction=True)    pipe.setex(key, 3600, json.dumps(data))    pipe.sadd(tag, key)    pipe.expire(tag, 3600)    pipe.execute()
def invalidate_category(category_id: int):    tag = f"tag:category:{category_id}"    keys = r.smembers(tag)    if keys:        r.delete(*keys)    r.delete(tag)

import timeimport uuid
def create_session(user_id: int, ttl: int = 86400) -> str:    session_id = str(uuid.uuid4())    key = f"session:{session_id}"    pipe = r.pipeline(transaction=True)    pipe.hset(key, mapping={        "user_id": user_id,        "created_at": int(time.time()),    })    pipe.expire(key, ttl)    pipe.execute()    return session_id
def get_session(session_id: str) -> dict | None:    data = r.hgetall(f"session:{session_id}")    return data if data else None
def delete_session(session_id: str):    r.delete(f"session:{session_id}")

def is_rate_limited(user_id: int, limit: int = 100, window: int = 60) -> bool:    key = f"ratelimit:{user_id}:{int(time.time()) // window}"    pipe = r.pipeline(transaction=True)    pipe.incr(key)    pipe.expire(key, window)    count, _ = pipe.execute()    return count > limit

sliding_window = r.register_script(open('sliding_window.lua').read())
def allow_request(user_id: int) -> bool:    key = f"ratelimit:sliding:{user_id}"    now = int(time.time() * 1000)    return bool(sliding_window(keys=[key], args=[now, 60000, 100]))

import uuid
def acquire_lock(resource: str, ttl_ms: int = 5000) -> str | None:    lock_key = f"lock:{resource}"    token = str(uuid.uuid4())    acquired = r.set(lock_key, token, px=ttl_ms, nx=True)    return token if acquired else None
def release_lock(resource: str, token: str) -> bool:    release_script = """    if redis.call('get', KEYS[1]) == ARGV[1] then        return redis.call('del', KEYS[1])    else        return 0    end    """    result = r.eval(release_script, 1, f"lock:{resource}", token)    return bool(result)
# Usagetoken = acquire_lock("order:payment:123")if token:    try:        process_payment()    finally:        release_lock("order:payment:123", token)

# Publisherdef publish_event(channel: str, payload: dict):    r.publish(channel, json.dumps(payload))
# Subscriber (blocking — run in separate thread/process)def subscribe_events(channel: str):    pubsub = r.pubsub()    pubsub.subscribe(channel)    for message in pubsub.listen():        if message['type'] == 'message':            handle(json.loads(message['data']))

# Producerdef emit(stream: str, event: dict):    r.xadd(stream, event, maxlen=10000)  # Cap stream length
# Consumer group — guarantees at-least-once deliverytry:    r.xgroup_create('events:orders', 'processor', id='0', mkstream=True)except Exception:    pass  # Group already exists
def consume(stream: str, group: str, consumer: str):    while True:        messages = r.xreadgroup(group, consumer, {stream: '>'}, count=10, block=2000)        for _, entries in (messages or []):            for msg_id, data in entries:                process(data)                r.xack(stream, group, msg_id)

from redis import ConnectionPool, Redis
pool = ConnectionPool(    host='localhost',    port=6379,    db=0,    max_connections=20,    decode_responses=True,    socket_connect_timeout=2,    socket_timeout=2,)
r = Redis(connection_pool=pool)

from redis.cluster import RedisCluster
r = RedisCluster(    startup_nodes=[{"host": "redis-1", "port": 6379}],    decode_responses=True,    skip_full_coverage_check=True,)

from redis.sentinel import Sentinel
sentinel = Sentinel(    [('sentinel-1', 26379), ('sentinel-2', 26379)],    socket_timeout=0.5,)master = sentinel.master_for('mymaster', decode_responses=True)replica = sentinel.slave_for('mymaster', decode_responses=True)

import threading
_locks: dict[str, threading.Lock] = {}_locks_mutex = threading.Lock()
def get_with_lock(key: str, fetch_fn, ttl: int = 300):    cached = r.get(key)    if cached:        return json.loads(cached)
    with _locks_mutex:        if key not in _locks:            _locks[key] = threading.Lock()        lock = _locks[key]    with lock:        cached = r.get(key)  # Re-check after acquiring lock        if cached:            return json.loads(cached)        value = fetch_fn()        r.setex(key, ttl, json.dumps(value))        return value