| 12345678910111213141516171819202122232425262728 |
- import aioredis
- from typing import Any, Optional
- import time
-
- class RedisSession:
- def __init__(self, redis_url: str, session_expiry: int = 3600):
- self.redis_url = redis_url
- self.session_expiry = session_expiry
- self.redis = None
-
- async def connect(self):
- self.redis = await aioredis.create_redis_pool(self.redis_url)
-
- async def disconnect(self):
- if self.redis:
- self.redis.close()
- await self.redis.wait_closed()
-
- async def get(self, key: str) -> Optional[Any]:
- value = await self.redis.get(key)
- return value.decode("utf-8") if value else None
-
- async def set(self, key: str, value: Any, expiry: int = None):
- expiry = expiry or self.session_expiry
- await self.redis.set(key, value, ex=expiry)
-
- async def delete(self, key: str):
- await self.redis.delete(key)
|
