redis update

This commit is contained in:
Untone 2023-10-13 13:13:45 +03:00
parent 82618bf7f3
commit bbd8f61408
2 changed files with 25 additions and 27 deletions

View File

@ -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

View File

@ -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()