diff --git a/requirements.txt b/requirements.txt index 6fd1749f..1af33c8f 100644 --- a/requirements.txt +++ b/requirements.txt @@ -3,7 +3,7 @@ git+https://github.com/encode/starlette.git#master git+https://github.com/graphql-python/gql.git#master SQLAlchemy uvicorn -aredis +redis[hiredis] itsdangerous Authlib PyJWT diff --git a/services/redis.py b/services/redis.py index a7c9fe0f..2afb24f0 100644 --- a/services/redis.py +++ b/services/redis.py @@ -1,58 +1,56 @@ -import asyncio -import aredis +import redis.asyncio as redis from settings import REDIS_URL - class RedisCache: def __init__(self, uri=REDIS_URL): self._uri: str = uri self.pubsub_channels = [] - self._instance = None + self._redis = None async def connect(self): - self._instance = aredis.StrictRedis.from_url(self._uri, decode_responses=True) + self._redis = redis.Redis.from_url(self._uri, decode_responses=True) async def disconnect(self): - if self._instance: - self._instance.connection_pool.disconnect() - self._instance = None + await self._redis.aclose() async def execute(self, command, *args, **kwargs): - while not self._instance: - await asyncio.sleep(1) + if not self._redis: + await self.connect() try: print("[redis] " + command + " " + " ".join(args)) - return await self._instance.execute_command(command, *args, **kwargs) - except Exception: - pass + return await self._redis.execute_command(command, *args, **kwargs) + except Exception as e: + print(f"[redis] error: {e}") + return None async def subscribe(self, *channels): - if not self._instance: + if not self._redis: await self.connect() - for channel in channels: - await self._instance.subscribe(channel) - self.pubsub_channels.append(channel) + async with self._redis.pubsub() as pubsub: + for channel in channels: + await pubsub.subscribe(channel) + self.pubsub_channels.append(channel) async def unsubscribe(self, *channels): - if not self._instance: + if not self._redis: return - for channel in channels: - await self._instance.unsubscribe(channel) - self.pubsub_channels.remove(channel) + async with self._redis.pubsub() as pubsub: + for channel in channels: + await pubsub.unsubscribe(channel) + self.pubsub_channels.remove(channel) async def publish(self, channel, data): - if not self._instance: + if not self._redis: return - await self._instance.publish(channel, data) + await self._redis.publish(channel, data) async def lrange(self, key, start, stop): print(f"[redis] LRANGE {key} {start} {stop}") - return await self._instance.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._instance.mget(key, *keys) - + return await self._redis.mget(key, *keys) redis = RedisCache()