database rewrite

tunebot/redis/__init__.py

@@ -0,0 +1,4 @@
+from tunebot.redis.autojoin import *
+from tunebot.redis.entity import *
+from tunebot.redis.playlist import *
+from tunebot.redis.playlist_source import *

tunebot/redis/autojoin.py

@@ -0,0 +1,41 @@
+from tunebot import AutoJoin
+from tunebot import GlobalAutoJoin
+from tunebot.redis import RedisBotEntity
+from tunebot.redis import RedisContextEntity
+
+
+class GlobalRedisAutoJoin(RedisBotEntity, GlobalAutoJoin):
+    async def fetch_channels(self) -> dict[str, list[str]]:
+        """
+        Retrieves all guilds (with their configurations) where AutoJoin is enabled
+        """
+        channels: dict[str, str] = await self.redis.hgetall(self.key("autojoin"))
+        return {key: value.split(":") for key, value in channels.items()}
+
+
+class RedisAutoJoin(RedisContextEntity, AutoJoin):
+    async def update(self, voice_channel_id: int, text_channel_id: int):
+        """
+        Upserts the configuration of an AutoJoin guild.
+
+        Args:
+            voice_channel_id (int): [description]
+            text_channel_id (int): [description]
+        """
+        if not self.ctx.guild:
+            raise Exception("This method can only be invoked inside of a guild.")
+
+        value = f"{voice_channel_id}:{text_channel_id}"
+        await self.redis.hset(self.key("autojoin"), self.ctx.guild.id, value)
+
+    async def disable(self):
+        """
+        Removes the AutoJoin configuration of a guild.
+        """
+        if not self.ctx.guild:
+            raise Exception("This method can only be invoked inside of a guild.")
+
+        await self.redis.hdel(self.key("autojoin"), self.ctx.guild.id)
+
+
+__all__ = ("GlobalRedisAutoJoin", "RedisAutoJoin")

tunebot/redis/entity.py

@@ -0,0 +1,49 @@
+from abc import ABC
+from abc import abstractmethod
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from context import CustomContext
+
+from aioredis.client import Redis
+
+
+class RedisEntity(ABC):
+    @property
+    @abstractmethod
+    def redis() -> Redis:
+        pass
+
+    @abstractmethod
+    def key(self, name: str) -> str:
+        pass
+
+
+class RedisBotEntity(RedisEntity):
+    def __init__(self, redis: Redis, prefix: str) -> None:
+        self._redis = redis
+        self.prefix = prefix
+        super().__init__()
+
+    @property
+    def redis(self) -> Redis:
+        return self._redis
+
+    def key(self, name: str) -> str:
+        return ":".join([self.prefix, name])
+
+
+class RedisContextEntity(RedisEntity):
+    def __init__(self, ctx: "CustomContext") -> None:
+        self.ctx = ctx
+        super().__init__()
+
+    @property
+    def redis(self) -> Redis:
+        return self.ctx.redis
+
+    def key(self, name: str) -> str:
+        return ":".join([self.ctx.bot.redis_prefix, name])
+
+
+__all__ = ("RedisEntity", "RedisBotEntity", "RedisContextEntity")

tunebot/redis/playlist.py

@@ -0,0 +1,33 @@
+from typing import Optional
+
+from tunebot.abc import GlobalPlaylist
+from tunebot.redis import RedisBotEntity
+
+
+class GlobalRedisPlaylist(RedisBotEntity, GlobalPlaylist):
+    async def pick_random(self, amount: Optional[int] = 1) -> set[str]:
+        """
+        Picks an amount of random tracks from the playlist in Redis
+
+        Args:
+            amount (Optional[int], optional): [description]. Defaults to 1.
+        """
+        return await self.redis.srandmember(self.key("playlist"), amount)
+
+    async def add_tracks(self, urls: list[str]):
+        """
+        Adds one or more urls to the playlist in Redis
+
+        Args:
+            urls (list[str]): [description]
+        """
+        await self.redis.sadd(self.key("playlist"), *urls)
+
+    async def clear(self):
+        """
+        Clears the entire playlist in Redis
+        """
+        await self.redis.delete(self.key("playlist"))
+
+
+__all__ = ("GlobalRedisPlaylist",)

tunebot/redis/playlist_source.py

@@ -0,0 +1,35 @@
+from tunebot import GlobalPlaylistSource
+from tunebot import PlaylistSource
+from tunebot.redis import RedisBotEntity
+from tunebot.redis import RedisContextEntity
+
+
+class GlobalRedisPlaylistSource(RedisBotEntity, GlobalPlaylistSource):
+    async def fetch_sources(self) -> set[str]:
+        """
+        Fetches all sources
+        """
+        return await self.redis.smembers(self.key("sources"))
+
+
+class RedisPlaylistSource(RedisContextEntity, PlaylistSource):
+    async def add(self, source_url: str):
+        """
+        Adds a playlist source to Redis
+
+        Args:
+            source_url (str): [description]
+        """
+        await self.redis.sadd(self.key("sources"), source_url)
+
+    async def remove(self, source_url: str) -> bool:
+        """
+        Removes a playlist source from Redis
+
+        Args:
+            source_url (str): [description]
+        """
+        return await self.redis.srem(self.key("sources"), source_url)
+
+
+__all__ = ("GlobalRedisPlaylistSource", "RedisPlaylistSource")