| 123456789101112131415161718192021222324252627282930313233343536373839 |
- import redis
- import os
- import uuid
- from datetime import timedelta
- # Redis Configuration (Environment variables for production)
- REDIS_HOST = os.getenv("REDIS_HOST", "localhost")
- REDIS_PORT = int(os.getenv("REDIS_PORT", 6379))
- REDIS_DB = int(os.getenv("REDIS_DB", 0))
- r = redis.Redis(host=REDIS_HOST, port=REDIS_PORT, db=REDIS_DB, decode_responses=True)
- def create_session(user_id: int, expires_days: int = 365) -> str:
- """Create a unique session ID in Redis and map it to user_id"""
- session_id = str(uuid.uuid4())
- # Save the session with an expiration time
- r.setex(f"session:{session_id}", timedelta(days=expires_days), str(user_id))
- return session_id
- def validate_session(session_id: str) -> bool:
- """Check if the session ID exists in Redis"""
- return r.exists(f"session:{session_id}") == 1
- def delete_session(session_id: str):
- """Delete a session from Redis (Logout)"""
- r.delete(f"session:{session_id}")
- def get_user_id_from_session(session_id: str):
- """Retrieve the user_id associated with a session"""
- return r.get(f"session:{session_id}")
- import time
- def track_user_ping(user_id: int):
- """Track user online status with 60s expiration"""
- r.setex(f"user_ping:{user_id}", 60, str(int(time.time())))
- def is_user_online(user_id: int) -> bool:
- """Check if user has pinged within the last 60s"""
- return r.exists(f"user_ping:{user_id}") == 1
|