shanxia2.0/backend/app/core/redis_client.py

"""共享 Redis 客户端（同步：会话鉴权；异步：API 加密 nonce）。"""
from __future__ import annotations

from typing import Optional

import redis
import redis.asyncio as aioredis

from app.core.config import settings

_sync_redis_client: Optional[redis.Redis] = None
_async_redis_client: Optional[aioredis.Redis] = None


def redis_enabled() -> bool:
    return bool(settings.REDIS_HOST)


def _redis_common_kwargs() -> dict:
    return {
        "host": settings.REDIS_HOST,
        "port": settings.REDIS_PORT,
        "db": settings.REDIS_DB,
        "password": settings.REDIS_PASSWORD or None,
        "socket_timeout": settings.REDIS_TIMEOUT_SEC,
        "socket_connect_timeout": settings.REDIS_CONNECT_TIMEOUT_SEC,
        "ssl": settings.REDIS_SSL_ENABLED,
        "decode_responses": True,
    }


def get_sync_redis_client() -> Optional[redis.Redis]:
    """同步 Redis（auth 会话等，在线程池中调用）。"""
    global _sync_redis_client
    if not redis_enabled():
        return None
    if _sync_redis_client is None:
        _sync_redis_client = redis.Redis(
            **_redis_common_kwargs(),
            health_check_interval=settings.REDIS_HEALTH_CHECK_INTERVAL_SEC,
        )
    return _sync_redis_client


def get_redis_client() -> Optional[aioredis.Redis]:
    """异步 Redis（API 加密 nonce 等中间件）。"""
    global _async_redis_client
    if not redis_enabled():
        return None
    if _async_redis_client is None:
        _async_redis_client = aioredis.Redis(
            **_redis_common_kwargs(),
            health_check_interval=settings.REDIS_HEALTH_CHECK_INTERVAL_SEC,
        )
    return _async_redis_client