feat: 完成redis的初始化动作

app/utils/redis_utils.py

@@ -0,0 +1,65 @@
+import uuid
+from contextlib import asynccontextmanager
+from typing import Annotated
+
+import redis.asyncio as redis
+from fastapi import Depends
+
+from app.config.env import env
+
+
+class RedisUtils():
+  def __init__(self):
+    self.redis_pool = None
+
+  # 检查 Redis 连接，先创建连接池管理对象
+  async def check_redis_connection(self):
+    self.redis_pool = redis.ConnectionPool(
+      host=env.redis_host,
+      port=env.redis_port,
+      password=env.redis_password,
+      db=env.redis_db,
+      encoding="utf-8",
+      decode_responses=True
+    )
+    # 立即对redis做一次读写测试
+    async with self.get_redis_connection() as redis_client:
+      try:
+        await redis_client.ping()
+        random_id = str(uuid.uuid4())
+        await redis_client.set("__init__setup__", random_id)
+        cache_random_id = await  redis_client.get("__init__setup__")
+        if cache_random_id != random_id:
+          raise Exception("Redis 连接失败，读写测试不一致")
+        print("✅ Redis connection successful：", f"redis://{env.redis_password}@${env.redis_host}:{env.redis_port}/{env.redis_db}")
+      except Exception as e:
+        print(f"❌ Database connection failed")
+        raise e
+
+  # 获取一个Redis连接客户端
+  @asynccontextmanager
+  async def get_redis_connection(self):
+    if not self.redis_pool:
+      raise Exception("Redis 连接池未初始化")
+    redis_client = redis.Redis(connection_pool=self.redis_pool)
+    try:
+      yield redis_client
+    finally:
+      await redis_client.close()
+
+  # 清理Redis连接池
+  async def close_redis_connection(self):
+    if self.redis_pool:
+      await self.redis_pool.disconnect()
+
+
+redis_utils = RedisUtils()
+
+
+# 作用：用于在接口中注入得到会话实例对象session，在接口执行完毕之后，自动执行close动作关闭会话
+async def get_redis_client() -> redis.Redis:
+  async with redis_utils.get_redis_connection() as redis_client:
+    yield redis_client
+
+
+RedisClientDep = Annotated[redis.Redis, Depends(get_redis_client)]