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

redis-multi-exec-fix-7

Browse Source
This commit is contained in:
Untone 2023-10-13 12:40:33 +03:00
parent 29478aa04b
commit a148e1b1ab
1 changed files with 14 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

28
services/redis.py
View File

@ -1,56 +1,56 @@
import redis.asyncio as redis import redis.asyncio as aredis
from settings import REDIS_URL from settings import REDIS_URL
class RedisCache: class RedisCache:
def __init__(self, uri=REDIS_URL): def __init__(self, uri=REDIS_URL):
self._uri: str = uri self._uri: str = uri
self.pubsub_channels = [] self.pubsub_channels = []
self._redis = None self._client = None
async def connect(self): async def connect(self):
self._redis = redis.Redis.from_url(self._uri, decode_responses=True) self._client = aredis.Redis.from_url(self._uri, decode_responses=True)
async def disconnect(self): async def disconnect(self):
await self._redis.aclose() await self._client.aclose()
async def execute(self, command, *args, **kwargs): async def execute(self, command, *args, **kwargs):
if not self._redis: if not self._client:
await self.connect() await self.connect()
try: try:
print("[redis] " + command + " " + " ".join(args)) print("[redis] " + command + " " + " ".join(args))
return await self._redis.execute_command(command, *args, **kwargs) return await self._client.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 return None
async def subscribe(self, *channels): async def subscribe(self, *channels):
if not self._redis: if not self._client:
await self.connect() await self.connect()
async with self._redis.pubsub() as pubsub: async with self._client.pubsub() as pubsub:
for channel in channels: for channel in channels:
await pubsub.subscribe(channel) await 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._client:
return return
async with self._redis.pubsub() as pubsub: async with self._client.pubsub() as pubsub:
for channel in channels: for channel in channels:
await pubsub.unsubscribe(channel) await 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):
if not self._redis: if not self._client:
return return
await self._redis.publish(channel, data) await self._client.publish(channel, data)
async def lrange(self, key, start, stop): async def lrange(self, key, start, stop):
print(f"[redis] LRANGE {key} {start} {stop}") print(f"[redis] LRANGE {key} {start} {stop}")
return await self._redis.lrange(key, start, stop) return await self._client.lrange(key, start, stop)
async def mget(self, key, *keys): async def mget(self, key, *keys):
print(f"[redis] MGET {key} {keys}") print(f"[redis] MGET {key} {keys}")
return await self._redis.mget(key, *keys) return await self._client.mget(key, *keys)
redis = RedisCache() redis = RedisCache()