session_utils.py 1.4 KB

123456789101112131415161718192021222324252627282930313233343536373839
  1. import redis
  2. import os
  3. import uuid
  4. from datetime import timedelta
  5. # Redis Configuration (Environment variables for production)
  6. REDIS_HOST = os.getenv("REDIS_HOST", "localhost")
  7. REDIS_PORT = int(os.getenv("REDIS_PORT", 6379))
  8. REDIS_DB = int(os.getenv("REDIS_DB", 0))
  9. r = redis.Redis(host=REDIS_HOST, port=REDIS_PORT, db=REDIS_DB, decode_responses=True)
  10. def create_session(user_id: int, expires_days: int = 365) -> str:
  11. """Create a unique session ID in Redis and map it to user_id"""
  12. session_id = str(uuid.uuid4())
  13. # Save the session with an expiration time
  14. r.setex(f"session:{session_id}", timedelta(days=expires_days), str(user_id))
  15. return session_id
  16. def validate_session(session_id: str) -> bool:
  17. """Check if the session ID exists in Redis"""
  18. return r.exists(f"session:{session_id}") == 1
  19. def delete_session(session_id: str):
  20. """Delete a session from Redis (Logout)"""
  21. r.delete(f"session:{session_id}")
  22. def get_user_id_from_session(session_id: str):
  23. """Retrieve the user_id associated with a session"""
  24. return r.get(f"session:{session_id}")
  25. import time
  26. def track_user_ping(user_id: int):
  27. """Track user online status with 60s expiration"""
  28. r.setex(f"user_ping:{user_id}", 60, str(int(time.time())))
  29. def is_user_online(user_id: int) -> bool:
  30. """Check if user has pinged within the last 60s"""
  31. return r.exists(f"user_ping:{user_id}") == 1