webmail-client/backend/app/redis_client.py

"""Shared async Redis client (sessions, rate limits, pub/sub, arq)."""

from __future__ import annotations

from collections.abc import Awaitable
from typing import Any, cast

import redis.asyncio as aioredis

from app.config import get_settings


def aw(result: Any) -> Awaitable[Any]:
    """Normalize a redis-py ``ResponseT`` (``Awaitable | value``) to an awaitable.

    Some redis-py async commands (hash ops, getdel, ...) are typed as a
    sync/async union; this keeps call sites awaitable under ``mypy --strict``.
    """
    return cast("Awaitable[Any]", result)


_client: aioredis.Redis | None = None
_bytes_client: aioredis.Redis | None = None


def get_redis() -> aioredis.Redis:
    """Return (creating on first use) the process-wide Redis client."""
    global _client
    if _client is None:
        _client = aioredis.Redis.from_url(
            get_settings().redis_url,
            encoding="utf-8",
            decode_responses=True,
        )
    return _client


def get_redis_bytes() -> aioredis.Redis:
    """Return the process-wide Redis client for binary values.

    ``get_redis()`` decodes every reply as UTF-8 text, which fails on (or
    silently changes the type of) raw bytes such as a cached image. Anything that
    stores bytes goes through this one instead.
    """
    global _bytes_client
    if _bytes_client is None:
        _bytes_client = aioredis.Redis.from_url(get_settings().redis_url, decode_responses=False)
    return _bytes_client


async def close_redis() -> None:
    """Close the Redis clients (used on app shutdown)."""
    global _client, _bytes_client
    if _client is not None:
        await _client.aclose()
        _client = None
    if _bytes_client is not None:
        await _bytes_client.aclose()
        _bytes_client = None