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

aredid

Browse Source
This commit is contained in:
Tony Rewin 2023-10-05 00:20:43 +03:00
parent a04e81695d
commit 0de9270af9
3 changed files with 12 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
requirements.txt
View File

@ -1,5 +1,5 @@
sentry-sdk sentry-sdk
aioredis aredis
ariadne ariadne
starlette starlette
uvicorn uvicorn

2
services/presence.py
View File

@ -6,6 +6,6 @@ async def notify_message(message, chat_id: str):
channel_name = f"chat:{chat_id}" channel_name = f"chat:{chat_id}"
data = {**message, "kind": "new_message"} data = {**message, "kind": "new_message"}
try: try:
await redis.execute_pubsub("PUBLISH", channel_name, json.dumps(data)) await redis.publish(channel_name, json.dumps(data))
except Exception as e: except Exception as e:
print(f"Failed to publish to channel {channel_name}: {e}") print(f"Failed to publish to channel {channel_name}: {e}")

27
services/redis.py
View File

@ -1,5 +1,5 @@
import asyncio import asyncio
import aioredis import aredis
from settings import REDIS_URL from settings import REDIS_URL
@ -10,13 +10,10 @@ class RedisCache:
self._redis = None self._redis = None
async def connect(self): async def connect(self):
pool = aioredis.ConnectionPool.from_url( self._redis = aredis.StrictRedis.from_url(self._uri, decode_responses=True)
self._uri, encoding="utf-8", max_connections=10
)
self._redis = aioredis.Redis(connection_pool=pool)
async def disconnect(self): async def disconnect(self):
await self._redis.wait_closed() self._redis.connection_pool.disconnect()
self._redis = None self._redis = None
async def execute(self, command, *args, **kwargs): async def execute(self, command, *args, **kwargs):
@ -28,29 +25,25 @@ class RedisCache:
except Exception: except Exception:
pass pass
async def execute_pubsub(self, command, *args, **kwargs):
while not self._redis:
await asyncio.sleep(1)
try:
print("[redis] " + command + " " + " ".join(args))
return await self._redis.execute_pubsub(command, *args, **kwargs)
except Exception:
pass
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.execute_pubsub("SUBSCRIBE", channel) await self._redis.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.execute_pubsub("UNSUBSCRIBE", channel) await self._redis.unsubscribe(channel)
self.pubsub_channels.remove(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): 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._redis.lrange(key, start, stop)