backend/app/database/redis.py

from __future__ import annotations

from redis.asyncio import Redis

from app.configs.settings import Settings


class RedisPool:
    def __init__(self, settings: Settings) -> None:
        self._settings = settings
        self._client: Redis | None = None

    async def connect(self) -> None:
        self._client = Redis.from_url(self._settings.redis.url, decode_responses=True)
        await self._client.ping()

    async def disconnect(self) -> None:
        if self._client is not None:
            await self._client.aclose()
        self._client = None

    @property
    def client(self) -> Redis:
        if self._client is None:
            msg = "Redis is not connected."
            raise RuntimeError(msg)
        return self._client