封装 redis 客户端处理方案

app/core/lifespan.py

@@ -9,7 +9,7 @@ from app.core.logger import log
 async def lifespan(app: FastAPI):
     """应用生命周期管理：初始化和释放所有连接资源"""
     from app.core.database import init_db, close_db
-    from app.core.redis import init_redis, close_redis
+    from app.core.redis import RedisManager
     from pathlib import Path
 
     # 启动：初始化资源
@@ -22,7 +22,7 @@ async def lifespan(app: FastAPI):
 
     try:
         log.info("初始化redis连接")
-        await init_redis()
+        await RedisManager.init()
     except Exception as e:
         log.warning(f"Redis 连接初始化失败: {e}")
         raise e
@@ -39,5 +39,5 @@ async def lifespan(app: FastAPI):
 
     # 关闭：释放资源
     await close_db()
-    await close_redis()
+    await RedisManager.close()
     log.info("所有连接资源已释放")

app/core/middleware.py

@@ -78,12 +78,12 @@ class JwtAuthMiddleware(BaseHTTPMiddleware):
         if not user_id or not uu_id:
             return
 
-        from app.core.redis import redis_client
+        from app.core.redis import RedisManager
-        if redis_client is None:
+        if RedisManager.client is None:
             return
 
         redis_key = f"client:login:token:{user_id}"
-        raw = await redis_client.get(redis_key)
+        raw = await RedisManager.client.get(redis_key)
         if not raw:
             return
 

app/core/redis.py

@@ -1,3 +1,9 @@
+"""Redis 异步连接管理
+
+通过 RedisManager 类管理 Redis 连接，使用类属性持有客户端实例。
+所有模块通过 from import RedisManager 导入后，访问 RedisManager.client 即可获取最新连接。
+"""
+
 from typing import Optional
 
 import redis.asyncio as aioredis
@@ -5,35 +11,34 @@ import redis.asyncio as aioredis
 from app.config import settings
 from app.core.logger import log
 
-redis_client: Optional[aioredis.Redis] = None
 
+class RedisManager:
+    """Redis 连接管理器（纯静态类，禁止实例化）"""
 
-async def init_redis() -> None:
+    client: Optional[aioredis.Redis] = None
-    """初始化 Redis 连接池"""
-    global redis_client
-    try:
-        redis_client = aioredis.from_url(
-            settings.redis_url,
-            max_connections=settings.redis_pool_size,
-            decode_responses=True,
-        )
-        await redis_client.ping()
-        log.info("Redis 连接池已初始化")
-    except Exception as e:
-        log.error(f"Redis 连接初始化失败: {e}")
-        raise
 
+    def __init__(self):
+        raise TypeError("RedisManager 是静态工具类，禁止实例化")
 
-async def close_redis() -> None:
+    @classmethod
-    """关闭 Redis 连接池"""
+    async def init(cls) -> None:
-    global redis_client
+        """初始化 Redis 连接池"""
-    if redis_client:
+        try:
-        await redis_client.close()
+            cls.client = aioredis.from_url(
-        log.info("Redis 连接池已关闭")
+                settings.redis_url,
+                max_connections=settings.redis_pool_size,
+                decode_responses=True,
+            )
+            await cls.client.ping()
+            log.info("Redis 连接池已初始化")
+        except Exception as e:
+            log.error(f"Redis 连接初始化失败: {e}")
+            raise
 
-
+    @classmethod
-async def get_redis() -> aioredis.Redis:
+    async def close(cls) -> None:
-    """依赖注入：提供 Redis 客户端实例"""
+        """关闭 Redis 连接池"""
-    if redis_client is None:
+        if cls.client:
-        raise RuntimeError("Redis 未初始化，请先调用 init_redis()")
+            await cls.client.close()
-    return redis_client
+            cls.client = None
+            log.info("Redis 连接池已关闭")

app/services/skill_gap_service.py

@@ -20,7 +20,7 @@ from app.ai.skill_gap_analyzer.analyzer import (
 )
 from app.ai.skill_gap_analyzer.prompts import MODULE_SCHEMAS
 from app.core.logger import log
-from app.core.redis import redis_client
+from app.core.redis import RedisManager
 from app.schemas.skill_gap import (
     CustomizeResume, ResumeProfile, Education, Work, Internship, Project, Competition, Paragraph,
 )
@@ -177,10 +177,10 @@ class SkillGapService:
     async def get_customize_resume(self, user_id: int) -> dict | None:
         """查询定制简历"""
         key = f"{CUSTOMIZE_RESUME_KEY_PREFIX}{user_id}"
-        data = await redis_client.get(key)
+        data = await RedisManager.client.get(key)
-        if not data:
+        if data:
-            return None
+            return CustomizeResume.model_validate_json(data).model_dump(by_alias=True)
-        return CustomizeResume.model_validate_json(data).model_dump(by_alias=True)
+        return None
 
     async def update_customize_resume(self, user_id: int, data: dict) -> None:
         """手动编辑定制简历（整体覆盖）"""
@@ -190,12 +190,12 @@ class SkillGapService:
     async def rollback_customize_resume(self, user_id: int) -> None:
         """回滚定制简历"""
         rollback_key = f"{CUSTOMIZE_RESUME_ROLLBACK_KEY_PREFIX}{user_id}"
-        data = await redis_client.get(rollback_key)
+        data = await RedisManager.client.get(rollback_key)
         if not data:
             raise ValueError("没有可回滚的版本")
         key = f"{CUSTOMIZE_RESUME_KEY_PREFIX}{user_id}"
-        await redis_client.set(key, data, ex=CUSTOMIZE_RESUME_EXPIRE)
+        await RedisManager.client.set(key, data, ex=CUSTOMIZE_RESUME_EXPIRE)
-        await redis_client.delete(rollback_key)
+        await RedisManager.client.delete(rollback_key)
 
     # ===== AI 对话编辑 =====
 
@@ -204,7 +204,7 @@ class SkillGapService:
         """AI 对话式编辑定制简历"""
         # 1. 取当前定制简历
         key = f"{CUSTOMIZE_RESUME_KEY_PREFIX}{user_id}"
-        raw = await redis_client.get(key)
+        raw = await RedisManager.client.get(key)
         if not raw:
             raise ValueError("定制简历不存在，请先生成")
         cr = CustomizeResume.model_validate_json(raw)
@@ -241,7 +241,7 @@ class SkillGapService:
             self._apply_edit_result(cr, mod_key, result)
         # 6. 保存回滚 + 新版本
         rollback_key = f"{CUSTOMIZE_RESUME_ROLLBACK_KEY_PREFIX}{user_id}"
-        await redis_client.set(rollback_key, raw, ex=CUSTOMIZE_RESUME_ROLLBACK_EXPIRE)
+        await RedisManager.client.set(rollback_key, raw, ex=CUSTOMIZE_RESUME_ROLLBACK_EXPIRE)
         await self._save_customize_resume(user_id, cr)
         label = plan.get("updatedModulesLabel", "简历内容")
         return {"type": "updated", "message": f"完成！已更新：{label}"}
@@ -352,4 +352,4 @@ class SkillGapService:
     async def _save_customize_resume(user_id: int, cr: CustomizeResume) -> None:
         """存定制简历到 Redis"""
         key = f"{CUSTOMIZE_RESUME_KEY_PREFIX}{user_id}"
-        await redis_client.set(key, cr.model_dump_json(by_alias=True), ex=CUSTOMIZE_RESUME_EXPIRE)
+        await RedisManager.client.set(key, cr.model_dump_json(by_alias=True), ex=CUSTOMIZE_RESUME_EXPIRE)