File size: 2,053 Bytes
a74b879 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 | import os
import json
import redis
from datetime import timedelta
REDIS_URL = os.environ.get('REDIS_URL', 'redis://localhost:6379/0')
# Initialize Production Redis Pool
_redis_pool = redis.ConnectionPool.from_url(REDIS_URL, max_connections=50, decode_responses=True)
redis_client = redis.Redis(connection_pool=_redis_pool)
def get(key, default=None):
"""Retrieve JSON-deserialized object from Redis"""
try:
val = redis_client.get(key)
return json.loads(val) if val else default
except Exception:
# Silently fail for local dev if Redis not present
return default
def set(key, value, ttl_seconds=3600):
"""Store JSON object in Redis with expiration"""
try:
val = json.dumps(value, ensure_ascii=False)
redis_client.setex(key, timedelta(seconds=ttl_seconds), val)
return True
except Exception:
return False
def check_rate_limit(user_id: int, plan: str, resource: str) -> bool:
"""Enterprise Rate Limiting Strategy (Token Bucket/Counter per User Plan)"""
limits = {
'free': 10,
'pro': 100,
'enterprise': 1000
}
limit = limits.get(plan.lower(), 10)
# Key strategy: rate_limit:crawls:user_123:minute
import time
current_minute = int(time.time() / 60)
key = f"rate_limit:{resource}:user_{user_id}:{current_minute}"
try:
pipe = redis_client.pipeline()
pipe.incr(key, 1)
pipe.expire(key, 60) # reset window next minute
results = pipe.execute()
current_usage = results[0]
return current_usage <= limit
except Exception:
return True # Fail open to avoid blocking users if Redis drops
def acquire_lock(lock_key: str, timeout: int = 10) -> bool:
"""Implement Idempotency and Job Deduplication using Redis Locks"""
try:
# returns True if set was successful (lock acquired), False if existing
return bool(redis_client.set(lock_key, "locked", nx=True, ex=timeout))
except Exception as e:
return False
|