import asyncio import aredis from settings import REDIS_URL class RedisCache: def __init__(self, uri=REDIS_URL): self._uri: str = uri self.pubsub_channels = [] self._redis = None async def connect(self): self._redis = aredis.StrictRedis.from_url(self._uri, decode_responses=True) async def disconnect(self): self._redis.connection_pool.disconnect() self._redis = None async def execute(self, command, *args, **kwargs): while not self._redis: await asyncio.sleep(1) try: print("[redis] " + command + " " + " ".join(args)) return await self._redis.execute_command(command, *args, **kwargs) except Exception: pass async def subscribe(self, *channels): if not self._redis: await self.connect() for channel in channels: await self._redis.subscribe(channel) self.pubsub_channels.append(channel) async def unsubscribe(self, *channels): if not self._redis: return for channel in channels: await self._redis.unsubscribe(channel) self.pubsub_channels.remove(channel) async def publish(self, channel, data): if not self._redis: return await self._redis.publish(channel, data) async def lrange(self, key, start, stop): print(f"[redis] LRANGE {key} {start} {stop}") return await self._redis.lrange(key, start, stop) async def mget(self, key, *keys): print(f"[redis] MGET {key} {keys}") return await self._redis.mget(key, *keys) async def pipeline(self): print(f"[redis] pipeline") return await self._redis.pipeline() redis = RedisCache() __all__ = ["redis"]