Redis #

pip install redis

# Docker (easiest)
docker run -d --name redis -p 6379:6379 redis:latest

# With a password
docker run -d --name redis -p 6379:6379 redis:latest redis-server --requirepass "secret"

import redis
import os

# ANTI-PATTERN: hardcoded connection without password, without decode_responses
r = redis.Redis(host="localhost", port=6379, db=0)
nilai = r.get("kunci")   # ✗ -- the value is bytes: b"data", not a string

# CORRECT: use environment variables, password, and decode_responses=True
r = redis.Redis(
    host=os.getenv("REDIS_HOST", "localhost"),
    port=int(os.getenv("REDIS_PORT", "6379")),
    password=os.getenv("REDIS_PASSWORD"),
    db=int(os.getenv("REDIS_DB", "0")),
    decode_responses=True,    # return str instead of bytes
    socket_timeout=5,         # timeout during operations
    socket_connect_timeout=5  # timeout when connecting
)

# Or use a URL (more concise)
r = redis.from_url(
    os.getenv("REDIS_URL", "redis://localhost:6379/0"),
    decode_responses=True,
    socket_timeout=5
)

# Test the connection
try:
    r.ping()
    print("Redis connection successful.")
except redis.ConnectionError as e:
    print(f"Connection failed: {e}")

# Connection Pool -- one pool for the whole application
pool = redis.ConnectionPool(
    host=os.getenv("REDIS_HOST", "localhost"),
    port=int(os.getenv("REDIS_PORT", "6379")),
    password=os.getenv("REDIS_PASSWORD"),
    db=0,
    decode_responses=True,
    max_connections=20    # connection limit in the pool
)

# Create a client that uses the pool
r = redis.Redis(connection_pool=pool)

# All operations use connections from the same pool
# Connections are automatically returned to the pool after an operation finishes

import json
from datetime import timedelta

# Basic Set and Get
r.set("nama", "Budi Santoso")
print(r.get("nama"))   # "Budi Santoso"

# Set with TTL (Time To Live) -- the key is automatically deleted after N seconds
r.set("token:abc123", "user-42", ex=3600)          # expire in 3600 seconds (1 hour)
r.set("otp:081234", "7823", px=300000)             # expire in 300000 milliseconds (5 minutes)
r.setex("session:xyz", timedelta(hours=24), "data")  # use a timedelta

# Check the remaining TTL
print(r.ttl("token:abc123"))   # remaining seconds, -1 if no TTL, -2 if the key doesn't exist

# Increment / decrement -- atomic
r.set("halaman_dilihat", 0)
r.incr("halaman_dilihat")         # +1
r.incrby("halaman_dilihat", 10)   # +10
r.decr("halaman_dilihat")         # -1
print(r.get("halaman_dilihat"))   # "9"

# Store JSON (manual serialization)
pengguna = {"id": 1, "nama": "Budi", "email": "[email protected]"}
r.set("pengguna:1", json.dumps(pengguna), ex=3600)

# Fetch JSON
data_raw = r.get("pengguna:1")
pengguna = json.loads(data_raw) if data_raw else None

# Atomic operation: SET only if the key doesn't exist (NX = Not eXists)
berhasil = r.set("kunci_unik", "nilai", nx=True, ex=60)
print("Set successful:", berhasil)   # True if newly set, None if it already exists

import json
from typing import Callable, Any

def get_dengan_cache(
    r:          redis.Redis,
    cache_key:  str,
    fetch_fn:   Callable,       # function to fetch data from the source (DB, API, etc.)
    ttl_detik:  int = 3600
) -> Any:
    """
    Cache-Aside pattern:
    1. Check the cache
    2. If hit → return from the cache
    3. If miss → fetch from the source, store in the cache, return
    """
    # Check the cache
    cached = r.get(cache_key)
    if cached is not None:
        print(f"Cache HIT: {cache_key}")
        return json.loads(cached)

    # Cache miss -- fetch from the source
    print(f"Cache MISS: {cache_key}")
    data = fetch_fn()

    if data is not None:
        r.set(cache_key, json.dumps(data, ensure_ascii=False), ex=ttl_detik)

    return data

# Usage example
def ambil_produk_dari_db(produk_id: int) -> dict:
    # Simulate a database query
    return {"id": produk_id, "nama": "Laptop Gaming", "harga": 18500000}

produk = get_dengan_cache(
    r,
    cache_key=f"produk:{101}",
    fetch_fn=lambda: ambil_produk_dari_db(101),
    ttl_detik=1800
)
print(produk)

def update_produk(r: redis.Redis, produk_id: int, data_baru: dict) -> None:
    """Update the database and invalidate related caches."""
    # 1. Update the database (not shown)
    # update_produk_di_db(produk_id, data_baru)

    # 2. Delete the stale cache
    r.delete(f"produk:{produk_id}")

    # Or update the cache directly (write-through)
    r.set(
        f"produk:{produk_id}",
        json.dumps(data_baru, ensure_ascii=False),
        ex=1800
    )

def invalidasi_cache_pattern(r: redis.Redis, pattern: str) -> int:
    """
    Delete all keys matching a pattern.
    Use carefully in production -- SCAN can be slow on large data.
    """
    keys = r.keys(pattern)   # e.g., "produk:*", "session:user-42:*"
    if keys:
        return r.delete(*keys)
    return 0

# Delete all product caches
jumlah_dihapus = invalidasi_cache_pattern(r, "produk:*")
print(f"{jumlah_dihapus} cache keys deleted.")

# Set a hash
r.hset("pengguna:42", mapping={
    "nama":   "Budi Santoso",
    "email":  "[email protected]",
    "usia":   "28",
    "aktif":  "1"
})

# Get one field
nama = r.hget("pengguna:42", "nama")         # "Budi Santoso"

# Get all fields
semua = r.hgetall("pengguna:42")             # dict of all fields
print(semua)  # {'nama': 'Budi Santoso', 'email': '...', 'usia': '28', 'aktif': '1'}

# Get specific fields
nama, email = r.hmget("pengguna:42", ["nama", "email"])

# Update one field only (without changing the others)
r.hset("pengguna:42", "usia", "29")

# Delete one field
r.hdel("pengguna:42", "aktif")

# Set a TTL for the whole hash key
r.expire("pengguna:42", 3600)

# Increment a numeric field
r.hincrby("pengguna:42", "login_count", 1)

# Push to the right (FIFO queue)
r.rpush("antrian:email", json.dumps({"to": "[email protected]", "subject": "Welcome"}))
r.rpush("antrian:email", json.dumps({"to": "[email protected]", "subject": "Order"}))

# Pop from the left (get the first in)
item_raw = r.lpop("antrian:email")
item     = json.loads(item_raw) if item_raw else None

# Blocking pop -- wait until an item arrives (useful for workers)
item_raw = r.blpop("antrian:email", timeout=30)   # wait 30 seconds

# List length
print(r.llen("antrian:email"))

# Get all items without removing
semua = r.lrange("antrian:email", 0, -1)   # 0 = first index, -1 = last

# Trim the list to the last N items (sliding window)
r.ltrim("log:aktivitas", 0, 999)   # keep only the newest 1000 items

# Add to a set
r.sadd("tag:produk:101", "laptop", "gaming", "asus")
r.sadd("tag:produk:102", "laptop", "bisnis", "lenovo")

# Check membership
print(r.sismember("tag:produk:101", "gaming"))   # True
print(r.sismember("tag:produk:101", "bisnis"))   # False

# All members
print(r.smembers("tag:produk:101"))   # {'laptop', 'gaming', 'asus'}

# Set operations
irisan      = r.sinter("tag:produk:101", "tag:produk:102")   # {'laptop'}
gabungan    = r.sunion("tag:produk:101", "tag:produk:102")   # all tags
perbedaan   = r.sdiff("tag:produk:101", "tag:produk:102")    # {'gaming', 'asus'}

# Remove from a set
r.srem("tag:produk:101", "asus")

# Game score leaderboard
r.zadd("leaderboard:game", {
    "player_alice": 9500,
    "player_budi":  8750,
    "player_sari":  9200,
    "player_andi":  7800
})

# Add/update a score
r.zincrby("leaderboard:game", 500, "player_budi")   # +500 for budi

# Top 3 players (highest scores)
top3 = r.zrange("leaderboard:game", 0, 2, desc=True, withscores=True)
print("Top 3:")
for rank, (pemain, skor) in enumerate(top3, start=1):
    print(f"  #{rank} {pemain}: {int(skor)}")

# A player's rank (0-based)
rank = r.zrevrank("leaderboard:game", "player_budi")
print(f"Budi's rank: #{rank + 1}")

# A player's score
skor = r.zscore("leaderboard:game", "player_budi")
print(f"Budi's score: {int(skor)}")

def cek_rate_limit(r: redis.Redis, user_id: str, maks_request: int = 10, window_detik: int = 60) -> bool:
    """
    Sliding window rate limiter using a Sorted Set.
    Return True if the request is allowed, False if it exceeds the limit.
    """
    import time

    key     = f"rate_limit:{user_id}"
    sekarang = time.time()
    window_start = sekarang - window_detik

    pipe = r.pipeline()
    pipe.zremrangebyscore(key, 0, window_start)   # remove requests outside the window
    pipe.zadd(key, {str(sekarang): sekarang})     # add the current request
    pipe.zcard(key)                               # count requests in the window
    pipe.expire(key, window_detik)               # set a TTL so the key doesn't pile up
    results = pipe.execute()

    jumlah_request = results[2]
    return jumlah_request <= maks_request

# Check the rate limit
for i in range(12):
    diizinkan = cek_rate_limit(r, "user-42", maks_request=10, window_detik=60)
    print(f"Request {i+1}: {'✓ allowed' if diizinkan else '✗ rejected'}")

# ANTI-PATTERN: operations one by one (N network round-trips)
for i in range(100):
    r.set(f"kunci:{i}", f"nilai:{i}")   # ✗ -- 100 round-trips to Redis

# CORRECT: use a pipeline (1 round-trip for everything)
pipe = r.pipeline(transaction=False)   # transaction=False = faster, non-atomic
for i in range(100):
    pipe.set(f"kunci:{i}", f"nilai:{i}", ex=3600)
pipe.execute()   # ✓ -- one round-trip

# Pipeline with an atomic transaction (MULTI/EXEC)
with r.pipeline(transaction=True) as pipe:
    pipe.set("saldo:alice", 1000)
    pipe.set("saldo:budi",  500)
    pipe.execute()   # all or nothing

# Pipeline with WATCH (optimistic locking)
def transfer_saldo(r: redis.Redis, dari: str, ke: str, jumlah: int) -> bool:
    kunci_dari = f"saldo:{dari}"
    kunci_ke   = f"saldo:{ke}"

    with r.pipeline() as pipe:
        while True:
            try:
                pipe.watch(kunci_dari, kunci_ke)   # watch for changes
                saldo_dari = int(pipe.get(kunci_dari) or 0)

                if saldo_dari < jumlah:
                    pipe.unwatch()
                    return False

                pipe.multi()   # start the transaction
                pipe.decrby(kunci_dari, jumlah)
                pipe.incrby(kunci_ke,   jumlah)
                pipe.execute()   # commit -- fails if changed externally
                return True

            except redis.WatchError:
                # Data changed while we were processing -- try again
                continue

r.set("saldo:alice", 1000)
r.set("saldo:budi",  500)
berhasil = transfer_saldo(r, "alice", "budi", 300)
print(f"Transfer: {'successful' if berhasil else 'failed'}")
print(f"Alice: {r.get('saldo:alice')}, Budi: {r.get('saldo:budi')}")

import uuid
import time

def acquire_lock(r: redis.Redis, lock_name: str, ttl_detik: int = 30) -> str | None:
    """
    Acquire a distributed lock. Return lock_value on success, None on failure.
    Use SET NX EX -- atomic, can't be race-conditioned.
    """
    lock_key   = f"lock:{lock_name}"
    lock_value = str(uuid.uuid4())   # unique value per lock

    # SET only if it doesn't exist (NX), with a TTL (EX)
    berhasil = r.set(lock_key, lock_value, nx=True, ex=ttl_detik)
    return lock_value if berhasil else None

def release_lock(r: redis.Redis, lock_name: str, lock_value: str) -> bool:
    """
    Release a lock. Only the acquirer can release it (checks lock_value).
    Use a Lua script for atomicity.
    """
    lock_key = f"lock:{lock_name}"

    # Lua script: check the value then delete atomically
    lua_script = """
    if redis.call("get", KEYS[1]) == ARGV[1] then
        return redis.call("del", KEYS[1])
    else
        return 0
    end
    """
    result = r.eval(lua_script, 1, lock_key, lock_value)
    return bool(result)

# Using the distributed lock
def proses_dengan_lock(r: redis.Redis, operasi_id: str) -> None:
    lock_value = acquire_lock(r, f"operasi:{operasi_id}", ttl_detik=30)

    if not lock_value:
        print(f"Operation {operasi_id} is being processed by another instance, skipping.")
        return

    try:
        print(f"Lock acquired for operation {operasi_id}")
        # Process the critical operation here
        time.sleep(1)
        print(f"Operation {operasi_id} finished.")
    finally:
        release_lock(r, f"operasi:{operasi_id}", lock_value)
        print(f"Lock released for operation {operasi_id}")

proses_dengan_lock(r, "generate-report")

import threading

def publisher_loop(r: redis.Redis) -> None:
    """Publish messages to a Redis channel."""
    import time
    for i in range(5):
        payload = json.dumps({"event": "update", "seq": i})
        r.publish("notifikasi:order", payload)
        print(f"Published: {payload}")
        time.sleep(1)

def subscriber_loop(r: redis.Redis) -> None:
    """Subscribe to a channel and process messages."""
    pubsub = r.pubsub()
    pubsub.subscribe("notifikasi:order")

    for message in pubsub.listen():
        if message["type"] == "message":
            payload = json.loads(message["data"])
            print(f"Received: {payload}")

# Run the subscriber in a separate thread
sub_thread = threading.Thread(target=subscriber_loop, args=(r,), daemon=True)
sub_thread.start()

# Publisher on the main thread
publisher_loop(r)

from redis.exceptions import ConnectionError, TimeoutError, ResponseError

def get_cache_aman(r: redis.Redis, key: str) -> str | None:
    try:
        return r.get(key)
    except ConnectionError:
        print("Redis unreachable — falling back to the database.")
        return None
    except TimeoutError:
        print("Redis timeout — trying again.")
        return None
    except ResponseError as e:
        print(f"Redis response error: {e}")
        return None

# Pattern with a database fallback if Redis is down
def ambil_data(r: redis.Redis, key: str, fetch_fn) -> Any:
    try:
        cached = r.get(key)
        if cached:
            return json.loads(cached)
    except (ConnectionError, TimeoutError):
        pass   # Redis unavailable, go straight to the database

    data = fetch_fn()
    try:
        if data:
            r.set(key, json.dumps(data), ex=3600)
    except (ConnectionError, TimeoutError):
        pass   # Can't store the cache, but the data is still returned

    return data