zk
40 lines
1.0 KiB
Python
40 lines
1.0 KiB
Python
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
|