66 lines
2.0 KiB
Python
66 lines
2.0 KiB
Python
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
|
|
self._pubsub = None
|
|
self.loop = asyncio.get_event_loop()
|
|
|
|
async def connect(self):
|
|
self._redis = aredis.StrictRedis.from_url(self._uri, decode_responses=True, loop=self.loop)
|
|
await self._redis.connection_pool.get_connection()
|
|
self._pubsub = self._redis.pubsub()
|
|
response = await self.execute('PING')
|
|
print(f"[redis] PING response: {response}")
|
|
|
|
async def disconnect(self):
|
|
self._redis.connection_pool.re
|
|
self._redis = None
|
|
self._pubsub = 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 as e:
|
|
print(f"[redis] error: {e}")
|
|
raise
|
|
|
|
async def subscribe(self, *channels):
|
|
if not self._redis:
|
|
await self.connect()
|
|
for channel in channels:
|
|
await self._pubsub.subscribe(channel)
|
|
self.pubsub_channels.append(channel)
|
|
|
|
async def unsubscribe(self, *channels):
|
|
if not self._redis:
|
|
return
|
|
for channel in channels:
|
|
await self._pubsub.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)
|
|
|
|
redis = RedisCache()
|
|
|
|
__all__ = ["redis"]
|