初始话项目框架

app/core/redis.py

@@ -0,0 +1,39 @@
+from typing import Optional
+
+import redis.asyncio as aioredis
+
+from app.config import settings
+from app.core.logger import log
+
+redis_client: Optional[aioredis.Redis] = None
+
+
+async def init_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
+
+
+async def close_redis() -> None:
+    """关闭 Redis 连接池"""
+    global redis_client
+    if redis_client:
+        await redis_client.close()
+        log.info("Redis 连接池已关闭")
+
+
+async def get_redis() -> aioredis.Redis:
+    """依赖注入：提供 Redis 客户端实例"""
+    if redis_client is None:
+        raise RuntimeError("Redis 未初始化，请先调用 init_redis()")
+    return redis_client