discours.io/inbox
5
0
Fork 0
Code Pull Requests Actions Wiki Activity

redis-multi-exec-fix-4

Browse Source
This commit is contained in:
Untone 2023-10-13 11:57:28 +03:00
parent 9b1c0ff792
commit aee90edfef
1 changed files with 7 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
services/redis.py
View File

@ -8,15 +8,19 @@ class RedisCache:
self._uri: str = uri self._uri: str = uri
self.pubsub_channels = [] self.pubsub_channels = []
self._redis = None self._redis = None
self._pubsub = None
async def connect(self): async def connect(self):
self._redis = aredis.StrictRedis.from_url(self._uri, decode_responses=True) self._redis = aredis.StrictRedis.from_url(self._uri, decode_responses=True)
await self._redis.connection_pool.connect()
self._pubsub = self._redis.pubsub()
response = await self.execute('PING') response = await self.execute('PING')
print(f"[redis] PING response: {response}") print(f"[redis] PING response: {response}")
async def disconnect(self): async def disconnect(self):
self._redis.connection_pool.disconnect() self._redis.connection_pool.disconnect()
self._redis = None self._redis = None
self._pubsub = None
async def execute(self, command, *args, **kwargs): async def execute(self, command, *args, **kwargs):
while not self._redis: while not self._redis:
@ -26,20 +30,20 @@ class RedisCache:
return await self._redis.execute_command(command, *args, **kwargs) return await self._redis.execute_command(command, *args, **kwargs)
except Exception as e: except Exception as e:
print(f"[redis] error: {e}") print(f"[redis] error: {e}")
return None raise
async def subscribe(self, *channels): async def subscribe(self, *channels):
if not self._redis: if not self._redis:
await self.connect() await self.connect()
for channel in channels: for channel in channels:
await self._redis.subscribe(channel) await self._pubsub.subscribe(channel)
self.pubsub_channels.append(channel) self.pubsub_channels.append(channel)
async def unsubscribe(self, *channels): async def unsubscribe(self, *channels):
if not self._redis: if not self._redis:
return return
for channel in channels: for channel in channels:
await self._redis.unsubscribe(channel) await self._pubsub.unsubscribe(channel)
self.pubsub_channels.remove(channel) self.pubsub_channels.remove(channel)
async def publish(self, channel, data): async def publish(self, channel, data):
@ -55,8 +59,6 @@ class RedisCache:
print(f"[redis] MGET {key} {keys}") print(f"[redis] MGET {key} {keys}")
return await self._redis.mget(key, *keys) return await self._redis.mget(key, *keys)
redis = RedisCache() redis = RedisCache()
__all__ = ["redis"] __all__ = ["redis"]