mirror of
https://github.com/stardrophere/InsightRadar.git
synced 2026-06-05 23:07:51 +08:00
stardrophere
58 lines
1.7 KiB
Python
58 lines
1.7 KiB
Python
import logging
|
|
import os
|
|
from typing import Optional, TYPE_CHECKING
|
|
|
|
if TYPE_CHECKING:
|
|
from redis import Redis
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
try:
|
|
import redis # type: ignore
|
|
except ImportError: # pragma: no cover
|
|
redis = None # type: ignore
|
|
|
|
|
|
REDIS_URL = os.getenv("REDIS_URL", "").strip()
|
|
REDIS_CONNECT_TIMEOUT_SECONDS = float(os.getenv("REDIS_CONNECT_TIMEOUT_SECONDS", "2"))
|
|
REDIS_SOCKET_TIMEOUT_SECONDS = float(os.getenv("REDIS_SOCKET_TIMEOUT_SECONDS", "2"))
|
|
|
|
_redis_client: Optional["Redis"] = None
|
|
_initialized = False
|
|
|
|
|
|
def get_redis_client() -> Optional["Redis"]:
|
|
"""Return a singleton Redis client, or None when Redis is unavailable."""
|
|
global _redis_client, _initialized
|
|
|
|
if _initialized:
|
|
return _redis_client
|
|
|
|
_initialized = True
|
|
|
|
if not REDIS_URL:
|
|
logger.info("REDIS_URL 未配置，验证码将回退到数据库存储")
|
|
_redis_client = None
|
|
return _redis_client
|
|
|
|
if redis is None:
|
|
logger.warning("未安装 redis 包，验证码将回退到数据库存储")
|
|
_redis_client = None
|
|
return _redis_client
|
|
|
|
try:
|
|
_redis_client = redis.Redis.from_url(
|
|
REDIS_URL,
|
|
decode_responses=True,
|
|
socket_connect_timeout=REDIS_CONNECT_TIMEOUT_SECONDS,
|
|
socket_timeout=REDIS_SOCKET_TIMEOUT_SECONDS,
|
|
health_check_interval=30,
|
|
)
|
|
_redis_client.ping()
|
|
logger.info("Redis 连接成功，验证码将优先使用 Redis")
|
|
except Exception as exc: # pragma: no cover
|
|
logger.warning("Redis 连接失败，将回退到数据库存储。error=%s", exc)
|
|
_redis_client = None
|
|
|
|
return _redis_client
|