Redis দিয়ে ক্যাশিং, রেট লিমিটিং ও নির্ভরযোগ্যতা

mkdir -p ~/projects/bitbyte-weather && cd ~/projects/bitbyte-weather
uv init -q --bare --name bitbyte-weather
uv add -q "fastapi[standard]" redis
source .venv/bin/activate
# with pip instead of uv:
#   python -m venv .venv && source .venv/bin/activate
#   pip install -q "fastapi[standard]" redis

docker run -d --name bitbyte-redis -p 6379:6379 redis:8
# or the BSD-licensed fork:
docker run -d --name bitbyte-redis -p 6379:6379 valkey/valkey:9

python -c "import redis; print(redis.Redis.from_url('redis://localhost:6379').ping())"

True

import time


class TTLCache:
    """A tiny in-process cache: a dict whose entries expire."""

    def __init__(self, ttl_seconds, clock=time.monotonic):
        self.ttl, self.clock, self.data = ttl_seconds, clock, {}

    def get(self, key):
        item = self.data.get(key)
        if item is None:
            return None
        value, expires_at = item
        if self.clock() >= expires_at:
            del self.data[key]
            return None
        return value

    def set(self, key, value):
        self.data[key] = (value, self.clock() + self.ttl)


now = [1000.0]                                   # a fake clock we can move
cache = TTLCache(ttl_seconds=60, clock=lambda: now[0])
cache.set("weather:dhaka", {"temp_c": 31.5})
print(cache.get("weather:dhaka"))
now[0] += 61                                     # 61 seconds later
print(cache.get("weather:dhaka"))

{'temp_c': 31.5}
None

import redis

r = redis.Redis.from_url("redis://localhost:6379/0", decode_responses=True)
PREFIX = "bitbyte:p12:"


def clear_prefix(prefix=PREFIX):
    keys = list(r.scan_iter(match=prefix + "*", count=500))
    if keys:
        r.delete(*keys)
    return len(keys)


clear_prefix()
print(r.ping())
r.set(PREFIX + "demo:greeting", "hello", ex=60)                # ex = TTL in seconds
print(r.get(PREFIX + "demo:greeting"))
print(r.ttl(PREFIX + "demo:greeting") <= 60)
print(r.get(PREFIX + "demo:missing"))                          # a miss is None
print(r.incr(PREFIX + "demo:counter"), r.incr(PREFIX + "demo:counter"))  # atomic counter
print(r.set(PREFIX + "demo:lock", "worker-1", nx=True, ex=30))
print(r.set(PREFIX + "demo:lock", "worker-2", nx=True, ex=30))

True
hello
True
None
1 2
True
None

# file: main.py
"""BitByte Weather: a cached, rate-limited proxy in front of the Practice API."""
import json
import logging
import os
import uuid
from contextlib import asynccontextmanager

import httpx
import redis.asyncio as aioredis
from fastapi import Depends, FastAPI, HTTPException, Request, Response
from redis.exceptions import RedisError

PRACTICE_API_URL = os.environ.get("PRACTICE_API_URL", "http://localhost:8800")
REDIS_URL = os.environ.get("REDIS_URL", "redis://localhost:6379/0")
PREFIX = os.environ.get("KEY_PREFIX", "bitbyte:p12:")
WEATHER_TTL = 300   # seconds a cached forecast stays fresh
RATE_LIMIT = 5      # requests ...
RATE_WINDOW = 60    # ... per 60 seconds, per client

logger = logging.getLogger("weather")


@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.http = httpx.AsyncClient(
        base_url=PRACTICE_API_URL, timeout=httpx.Timeout(5.0, connect=2.0))
    app.state.redis = aioredis.Redis.from_url(
        REDIS_URL, decode_responses=True, socket_timeout=0.5, socket_connect_timeout=0.5)
    yield
    await app.state.http.aclose()
    await app.state.redis.aclose()


app = FastAPI(title="BitByte Weather", lifespan=lifespan)


async def count(request: Request, field: str) -> None:
    """Add 1 to a shared counter. Metrics must never break a request."""
    try:
        await request.app.state.redis.hincrby(PREFIX + "metrics", field, 1)
    except RedisError:
        pass


@app.middleware("http")
async def request_id_and_metrics(request: Request, call_next):
    request.state.request_id = request.headers.get("X-Request-ID") or uuid.uuid4().hex
    response = await call_next(request)
    response.headers["X-Request-ID"] = request.state.request_id
    await count(request, f"status:{response.status_code}")
    return response


def client_key(request: Request) -> str:
    """Who is calling: the API key if there is one, else the IP address."""
    return request.headers.get("X-API-Key") or request.client.host


async def rate_limit(request: Request, response: Response) -> None:
    """Fixed window: RATE_LIMIT requests per RATE_WINDOW seconds per client."""
    key = f"{PREFIX}rate:{client_key(request)}"
    try:
        async with request.app.state.redis.pipeline() as pipe:
            pipe.incr(key)
            pipe.expire(key, RATE_WINDOW, nx=True)   # the window starts at the first request
            pipe.ttl(key)
            used, _, ttl = await pipe.execute()
    except RedisError:
        logger.warning("rate limiter unavailable, allowing the request")   # fail open
        return
    remaining = max(0, RATE_LIMIT - used)
    headers = {"X-RateLimit-Limit": str(RATE_LIMIT), "X-RateLimit-Remaining": str(remaining)}
    if used > RATE_LIMIT:
        raise HTTPException(429, "Rate limit exceeded", headers={**headers, "Retry-After": str(ttl)})
    response.headers.update(headers)


@app.get("/health")
async def health(request: Request) -> dict:
    try:
        await request.app.state.redis.ping()
        redis_state = "up"
    except RedisError:
        redis_state = "down"
    return {"status": "ok", "redis": redis_state}


@app.get("/v1/weather/{city}", dependencies=[Depends(rate_limit)])
async def weather(city: str, request: Request, response: Response) -> dict:
    key = f"{PREFIX}weather:{city.lower()}"
    try:
        cached = await request.app.state.redis.get(key)
    except RedisError:
        cached = None
        response.headers["X-Cache"] = "BYPASS"   # degrade: no cache, still answer
    if cached is not None:
        response.headers["X-Cache"] = "HIT"
        await count(request, "cache:hit")
        return json.loads(cached)

    try:
        upstream = await request.app.state.http.get(
            "/v1/weather", params={"city": city},
            headers={"X-Request-ID": request.state.request_id})
    except httpx.TimeoutException:
        raise HTTPException(504, "Weather service timed out")
    except httpx.HTTPError:
        raise HTTPException(502, "Weather service unreachable")
    if upstream.status_code == 404:
        raise HTTPException(404, f"Unknown city: {city}")
    if upstream.status_code != 200:
        raise HTTPException(502, "Weather service error")

    data = upstream.json()
    response.headers.setdefault("X-Cache", "MISS")
    await count(request, "cache:miss")
    try:
        await request.app.state.redis.set(key, json.dumps(data), ex=WEATHER_TTL)
    except RedisError:
        pass
    return data


@app.delete("/v1/weather/{city}/cache", status_code=204)
async def forget_weather(city: str, request: Request) -> None:
    await request.app.state.redis.delete(f"{PREFIX}weather:{city.lower()}")


@app.get("/metrics")
async def metrics(request: Request) -> dict[str, int]:
    raw = await request.app.state.redis.hgetall(PREFIX + "metrics")
    return {name: int(value) for name, value in sorted(raw.items())}

export PRACTICE_API_URL=http://localhost:8800
export REDIS_URL=redis://localhost:6379/0
fastapi run main.py --port 8000 > server.log 2>&1 &
export SERVER_PID=$!
curl -s --retry 20 --retry-connrefused --retry-delay 0 http://localhost:8000/health > /dev/null

for i in 1 2 3; do
  curl -s -o /dev/null -w '%{http_code} X-Cache: %header{x-cache}\n' \
       -H "X-API-Key: nadia-key" http://localhost:8000/v1/weather/Dhaka
done
curl -s -H "X-API-Key: nadia-key" http://localhost:8000/v1/weather/dhaka

200 X-Cache: MISS
200 X-Cache: HIT
200 X-Cache: HIT
{"city":"Dhaka","temp_c":31.5,"humidity":78,"condition":"Humid"}

python - <<'EOF'
import os, redis
r = redis.Redis.from_url(os.environ["REDIS_URL"], decode_responses=True)
for name in sorted(r.scan_iter(match="bitbyte:p12:*")):
    print(name)
print(r.get("bitbyte:p12:weather:dhaka"))
EOF

bitbyte:p12:demo:counter
bitbyte:p12:demo:greeting
bitbyte:p12:demo:lock
bitbyte:p12:metrics
bitbyte:p12:rate:nadia-key
bitbyte:p12:weather:dhaka
{"city": "Dhaka", "temp_c": 31.5, "humidity": 78, "condition": "Humid"}

curl -s -o /dev/null -w '%{http_code}\n' -X DELETE -H "X-API-Key: tanvir-key" \
     http://localhost:8000/v1/weather/Dhaka/cache
curl -s -o /dev/null -w '%{http_code} X-Cache: %header{x-cache}\n' \
     -H "X-API-Key: tanvir-key" http://localhost:8000/v1/weather/Dhaka

204
200 X-Cache: MISS

import hashlib
import json

import httpx

PROMPT_VERSION = "faq-v3"
CHAT_URL = "http://localhost:8800/v1/chat"
HEADERS = {"Authorization": "Bearer token-nadia-001"}


def llm_cache_key(model, messages, temperature, user_scope=None):
    payload = {"model": model, "prompt_version": PROMPT_VERSION, "messages": messages,
               "temperature": temperature, "scope": user_scope}
    canonical = json.dumps(payload, sort_keys=True, separators=(",", ":"), ensure_ascii=False)
    return PREFIX + "llm:" + hashlib.sha256(canonical.encode()).hexdigest()[:32]


def cached_chat(messages, model="practice-llm-1", temperature=0.0, user_scope=None):
    if temperature > 0:                               # creative output: never cached
        return httpx.post(CHAT_URL, headers=HEADERS, timeout=10,
                          json={"model": model, "messages": messages}).json(), "SKIP"
    key = llm_cache_key(model, messages, temperature, user_scope)
    if (hit := r.get(key)) is not None:
        return json.loads(hit), "HIT"
    resp = httpx.post(CHAT_URL, headers=HEADERS, timeout=10,
                      json={"model": model, "messages": messages})
    resp.raise_for_status()                           # errors are not cached
    data = resp.json()
    r.set(key, json.dumps(data), ex=3600)
    return data, "MISS"


question = [{"role": "user", "content": "What is a vector database?"}]
saved = 0
for attempt in range(3):
    data, status = cached_chat(question)
    if status == "HIT":
        saved += data["usage"]["total_tokens"]
    print(status, data["message"]["content"])
print("tokens saved:", saved)
print(cached_chat([{"role": "user", "content": "what is a vector database"}])[1])
print(cached_chat(question, user_scope="user-42")[1])
print(cached_chat(question, temperature=0.7)[1])

MISS This is a practice answer about: What is a vector database. A real model would say more.
HIT This is a practice answer about: What is a vector database. A real model would say more.
HIT This is a practice answer about: What is a vector database. A real model would say more.
tokens saved: 44
MISS
MISS
SKIP

def fixed_window(key, limit, window, now):
    bucket = f"{PREFIX}fw:{key}:{int(now // window)}"
    with r.pipeline() as pipe:                        # MULTI/EXEC: both or neither
        pipe.incr(bucket)
        pipe.expire(bucket, window)
        used, _ = pipe.execute()
    return used <= limit


for t in [58, 58.5, 59, 59.5, 60, 60.5, 61, 61.5]:
    print(f"t={t:>4}: {'allowed' if fixed_window('nadia', 3, 60, t) else 'rejected'}")

t=  58: allowed
t=58.5: allowed
t=  59: allowed
t=59.5: rejected
t=  60: allowed
t=60.5: allowed
t=  61: allowed
t=61.5: rejected

def sliding_window(key, limit, window, now, request_id):
    zkey = f"{PREFIX}sw:{key}"
    with r.pipeline() as pipe:
        pipe.zremrangebyscore(zkey, 0, now - window)   # forget requests outside the window
        pipe.zadd(zkey, {request_id: now})
        pipe.zcard(zkey)
        pipe.expire(zkey, window)
        _, _, used, _ = pipe.execute()
    if used > limit:
        r.zrem(zkey, request_id)                       # a rejected request does not count
        return False
    return True


for n, t in enumerate([58, 58.5, 59, 59.5, 60, 60.5, 61, 61.5]):
    print(f"t={t:>4}: {'allowed' if sliding_window('nadia', 3, 60, t, f'req-{n}') else 'rejected'}")

t=  58: allowed
t=58.5: allowed
t=  59: allowed
t=59.5: rejected
t=  60: rejected
t=60.5: rejected
t=  61: rejected
t=61.5: rejected

TOKEN_BUCKET = r.register_script("""
local capacity = tonumber(ARGV[1])
local rate = tonumber(ARGV[2])
local now = tonumber(ARGV[3])
local cost = tonumber(ARGV[4])
local state = redis.call('HMGET', KEYS[1], 'tokens', 'ts')
local tokens = tonumber(state[1]) or capacity
local ts = tonumber(state[2]) or now
tokens = math.min(capacity, tokens + (now - ts) * rate)
local allowed = 0
if tokens >= cost then
  tokens = tokens - cost
  allowed = 1
end
redis.call('HSET', KEYS[1], 'tokens', tokens, 'ts', now)
redis.call('EXPIRE', KEYS[1], math.ceil(capacity / rate) + 1)
return {allowed, tostring(tokens)}
""")


def token_bucket(key, capacity, rate, now, cost=1):
    allowed, tokens = TOKEN_BUCKET(keys=[f"{PREFIX}tb:{key}"], args=[capacity, rate, now, cost])
    return bool(allowed), float(tokens)


for t in [0, 0, 0, 0, 0, 0, 2, 2, 2]:              # a burst of 6, then 2 seconds later 3 more
    ok, left = token_bucket("nadia", capacity=5, rate=1.0, now=t)
    print(f"t={t}: {'allowed' if ok else 'rejected'}  tokens left {left:g}")
print(token_bucket("nadia", capacity=5, rate=1.0, now=10, cost=4))   # an expensive request

t=0: allowed  tokens left 4
t=0: allowed  tokens left 3
t=0: allowed  tokens left 2
t=0: allowed  tokens left 1
t=0: allowed  tokens left 0
t=0: rejected  tokens left 0
t=2: allowed  tokens left 1
t=2: allowed  tokens left 0
t=2: rejected  tokens left 0
(True, 1.0)

for i in 5 6 7; do
  curl -s -o /dev/null -H "X-API-Key: nadia-key" http://localhost:8000/v1/weather/Dhaka \
       -w '%{http_code} remaining=%header{x-ratelimit-remaining} retry-after=%header{retry-after}\n'
done | sed -E 's/retry-after=[0-9]+/retry-after=N/'      # N = seconds left in her window
curl -s -H "X-API-Key: nadia-key" http://localhost:8000/v1/weather/Dhaka; echo
curl -s -o /dev/null -w 'tanvir: %{http_code}\n' -H "X-API-Key: tanvir-key" \
     http://localhost:8000/v1/weather/Dhaka

200 remaining=0 retry-after=
429 remaining=0 retry-after=N
429 remaining=0 retry-after=N
{"detail":"Rate limit exceeded"}
tanvir: 200

import asyncio

import httpx


async def call_upstream(client, limit, stats):
    async with limit:                                  # wait for a free slot
        stats["running"] += 1
        stats["peak"] = max(stats["peak"], stats["running"])
        resp = await client.get("/v1/slow", params={"seconds": 0.2})
        stats["running"] -= 1
        return resp.status_code


async def main():
    stats = {"running": 0, "peak": 0}
    limit = asyncio.Semaphore(3)                       # at most 3 calls in flight
    async with httpx.AsyncClient(base_url="http://localhost:8800", timeout=5) as client:
        codes = await asyncio.gather(*(call_upstream(client, limit, stats) for _ in range(10)))
    print(codes)
    print("peak concurrent calls:", stats["peak"])


asyncio.run(main())

[200, 200, 200, 200, 200, 200, 200, 200, 200, 200]
peak concurrent calls: 3

timeout = httpx.Timeout(5.0, connect=2.0)            # connect: 2 s; read, write, pool: 5 s
print(timeout)
try:
    httpx.get("http://localhost:8800/v1/slow", params={"seconds": 2}, timeout=0.5)
except httpx.TimeoutException as exc:
    print("gave up:", type(exc).__name__)

Timeout(connect=2.0, read=5.0, write=5.0, pool=5.0)
gave up: ReadTimeout

import time


def get_with_deadline(path, budget_seconds, max_attempts=4):
    deadline = time.monotonic() + budget_seconds
    for attempt in range(1, max_attempts + 1):
        remaining = deadline - time.monotonic()
        if remaining <= 0:
            return f"deadline exceeded after {attempt - 1} attempt(s)"
        try:
            resp = httpx.get("http://localhost:8800" + path, timeout=remaining)
        except httpx.TimeoutException:
            return f"deadline exceeded after {attempt} attempt(s)"
        if resp.status_code not in (429, 502, 503, 504):
            return f"{resp.status_code} on attempt {attempt}"
        time.sleep(min(0.1 * 2 ** attempt, max(0.0, deadline - time.monotonic())))
    return "gave up: too many attempts"


print(get_with_deadline("/v1/flaky?key=deadline-demo&fail_times=2", budget_seconds=3))
print(get_with_deadline("/v1/slow?seconds=5", budget_seconds=1))

200 on attempt 3
deadline exceeded after 1 attempt(s)

          failures >= threshold
 CLOSED ─────────────────────────▶ OPEN ──── calls fail fast, upstream untouched
   ▲                                 │
   │ trial call succeeds             │ cool-down elapsed
   │                                 ▼
   └─────────────────────────── HALF-OPEN ── one trial call allowed
                (trial fails ──▶ back to OPEN)

class CircuitOpenError(Exception):
    pass


class CircuitBreaker:
    def __init__(self, threshold=3, cooldown=30.0, clock=time.monotonic):
        self.threshold, self.cooldown, self.clock = threshold, cooldown, clock
        self.failures, self.opened_at = 0, None

    @property
    def state(self):
        if self.opened_at is None:
            return "closed"
        return "half-open" if self.clock() - self.opened_at >= self.cooldown else "open"

    def call(self, func, *args, **kwargs):
        if self.state == "open":
            raise CircuitOpenError("upstream marked as down; failing fast")
        try:
            result = func(*args, **kwargs)
        except Exception:
            self.failures += 1
            if self.state == "half-open" or self.failures >= self.threshold:
                self.opened_at = self.clock()          # (re)open
            raise
        self.failures, self.opened_at = 0, None        # success closes the circuit
        return result


def flaky_upstream():
    resp = httpx.get("http://localhost:8800/v1/flaky",
                     params={"key": "breaker-demo", "fail_times": 3}, timeout=2)
    resp.raise_for_status()
    return resp.json()


now = [0.0]
breaker = CircuitBreaker(threshold=3, cooldown=30, clock=lambda: now[0])
for step in range(1, 6):
    try:
        print(step, breaker.state, "->", breaker.call(flaky_upstream))
    except CircuitOpenError as exc:
        print(step, breaker.state, "->", exc)
    except httpx.HTTPStatusError as exc:
        print(step, breaker.state, "->", exc.response.status_code)
now[0] += 30                                           # the cool-down passes
print(6, breaker.state, "->", breaker.call(flaky_upstream))
print("state now:", breaker.state)

1 closed -> 503
2 closed -> 503
3 open -> 503
4 open -> upstream marked as down; failing fast
5 open -> upstream marked as down; failing fast
6 half-open -> {'ok': True, 'attempt': 4}
state now: closed

REDIS_URL=redis://localhost:6390/0 fastapi run main.py --port 8001 > degraded.log 2>&1 &
export DEGRADED_PID=$!
curl -s --retry 20 --retry-connrefused --retry-delay 0 http://localhost:8001/health; echo
curl -s -o /dev/null -w '%{http_code} X-Cache: %header{x-cache}\n' http://localhost:8001/v1/weather/Sylhet
kill $DEGRADED_PID

{"status":"ok","redis":"down"}
200 X-Cache: BYPASS

curl -s -o /dev/null -w 'X-Request-ID: %header{x-request-id}\n' \
     -H "X-Request-ID: nadia-debug-001" -H "X-API-Key: farhana-key" \
     http://localhost:8000/v1/weather/Khulna

X-Request-ID: nadia-debug-001

curl -s http://localhost:8000/metrics | python -m json.tool
kill $SERVER_PID

{
    "cache:hit": 5,
    "cache:miss": 3,
    "status:200": 9,
    "status:204": 1,
    "status:429": 3
}

# 1. Easy
def weather_cached(city):
    key = PREFIX + "ex:weather:" + city.lower()
    if (hit := r.get(key)) is not None:
        return json.loads(hit), "HIT"
    resp = httpx.get("http://localhost:8800/v1/weather", params={"city": city}, timeout=5)
    resp.raise_for_status()
    r.set(key, resp.text, ex=120)
    return resp.json(), "MISS"


print(weather_cached("Rajshahi"))
print(weather_cached("Rajshahi")[1])

({'city': 'Rajshahi', 'temp_c': 33.6, 'humidity': 58, 'condition': 'Sunny'}, 'MISS')
HIT

# 2. Medium
def weather_cached(city):
    key = PREFIX + "ex:weather:" + city.lower()
    if (hit := r.get(key)) is not None:
        return (None if hit == "__none__" else json.loads(hit)), "HIT"
    resp = httpx.get("http://localhost:8800/v1/weather", params={"city": city}, timeout=5)
    if resp.status_code == 404:
        r.set(key, "__none__", ex=30)                  # short TTL for "not found"
        return None, "MISS"
    resp.raise_for_status()
    r.set(key, resp.text, ex=120)
    return resp.json(), "MISS"


print(weather_cached("Atlantis"))
print(weather_cached("Atlantis"))

(None, 'MISS')
(None, 'HIT')

# 3. Hard
def llm_call_with_budget(user, prompt, now):
    estimate = len(prompt.split()) * 2
    ok, left = token_bucket(f"llm:{user}", capacity=1000, rate=10, now=now, cost=estimate)
    if not ok:
        return f"rejected (needs ~{estimate}, has {left:g})"
    resp = httpx.post(CHAT_URL, headers=HEADERS, timeout=10,
                      json={"messages": [{"role": "user", "content": prompt}]})
    used = resp.json()["usage"]["total_tokens"]
    return f"allowed (estimated {estimate}, used {used}, {left:g} left)"


prompt = " ".join(["word"] * 200)
for call in range(1, 6):
    print(call, llm_call_with_budget("nadia", prompt, now=0))
print("cleaned up keys:", clear_prefix() > 0)

1 allowed (estimated 400, used 412, 600 left)
2 allowed (estimated 400, used 412, 200 left)
3 rejected (needs ~400, has 200)
4 rejected (needs ~400, has 200)
5 rejected (needs ~400, has 200)
cleaned up keys: True