session_utils.py 1.1 KB

123456789101112131415161718192021222324252627282930
  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}")