redis

alien_gateway/config.py

@@ -1,5 +1,5 @@
 from pydantic_settings import BaseSettings, SettingsConfigDict
-from typing import List
+from typing import Any, Dict, List
 from dotenv import load_dotenv
 import os
 load_dotenv()
@@ -56,12 +56,30 @@ class Settings(BaseSettings):
         return nodes
 
     @property
+    def REDIS_SENTINEL_KWARGS(self) -> Dict[str, Any]:
+        kwargs: Dict[str, Any] = {}
+        if self.REDIS_SENTINEL_USERNAME:
+            kwargs["username"] = self.REDIS_SENTINEL_USERNAME
+        if self.REDIS_SENTINEL_PASSWORD:
+            kwargs["password"] = self.REDIS_SENTINEL_PASSWORD
+        return kwargs
+
+    @property
     def REDIS_SENTINEL_URL(self) -> str:
         # Celery sentinel transport 需要 "sentinel://host:port;sentinel://host:port"
         return ";".join(
             f"sentinel://{host}:{port}" for host, port in self.REDIS_SENTINEL_NODES
         )
 
+    @property
+    def REDIS_SENTINEL_TRANSPORT_OPTIONS(self) -> Dict[str, Any]:
+        return {
+            "master_name": self.REDIS_MASTER_NAME,
+            "sentinel_kwargs": self.REDIS_SENTINEL_KWARGS,
+            "password": self.REDIS_PASSWORD,
+            "db": self.REDIS_DB,
+        }
+
     model_config = SettingsConfigDict(
         case_sensitive=True,
         env_file=".env",

alien_util/celery_app.py

@@ -17,24 +17,8 @@ celery_app.conf.update(
     result_serializer="json",
     timezone="Asia/Shanghai",
     enable_utc=True,
-    broker_transport_options={
-        "master_name": settings.REDIS_MASTER_NAME,
-        "sentinel_kwargs": {
-            **({"username": settings.REDIS_SENTINEL_USERNAME} if settings.REDIS_SENTINEL_USERNAME else {}),
-            **({"password": settings.REDIS_SENTINEL_PASSWORD} if settings.REDIS_SENTINEL_PASSWORD else {}),
-        },
-        "password": settings.REDIS_PASSWORD,
-        "db": settings.REDIS_DB,
-    },
-    result_backend_transport_options={
-        "master_name": settings.REDIS_MASTER_NAME,
-        "sentinel_kwargs": {
-            **({"username": settings.REDIS_SENTINEL_USERNAME} if settings.REDIS_SENTINEL_USERNAME else {}),
-            **({"password": settings.REDIS_SENTINEL_PASSWORD} if settings.REDIS_SENTINEL_PASSWORD else {}),
-        },
-        "password": settings.REDIS_PASSWORD,
-        "db": settings.REDIS_DB,
-    },
+    broker_transport_options=settings.REDIS_SENTINEL_TRANSPORT_OPTIONS,
+    result_backend_transport_options=settings.REDIS_SENTINEL_TRANSPORT_OPTIONS,
     # 定时任务配置
     beat_schedule={
         "check-contract-expiry": {

alien_util/redis_client.py

@@ -7,12 +7,7 @@ from alien_gateway.config import settings
 
 
 def _sentinel_kwargs() -> dict:
-    kw = {}
-    if getattr(settings, "REDIS_SENTINEL_USERNAME", ""):
-        kw["username"] = settings.REDIS_SENTINEL_USERNAME
-    if getattr(settings, "REDIS_SENTINEL_PASSWORD", ""):
-        kw["password"] = settings.REDIS_SENTINEL_PASSWORD
-    return kw
+    return settings.REDIS_SENTINEL_KWARGS
 
 
 def get_sentinel_client() -> Sentinel:
@@ -61,3 +56,13 @@ def get_async_redis_slave() -> AsyncRedis:
         password=settings.REDIS_PASSWORD or None,
         db=settings.REDIS_DB,
     )
+
+
+def get_redis() -> Redis:
+    # 默认返回主节点客户端，统一写入入口
+    return get_redis_master()
+
+
+def get_async_redis() -> AsyncRedis:
+    # 默认返回主节点客户端，统一写入入口
+    return get_async_redis_master()