Redis Python异步客户端莫名要求凭证问题求助

针对你遇到的NotImplementedError: get_credentials must be implemented错误，以及异步连接池被替换为同步池的问题，可按以下步骤排查修复：

1. 强制使用异步连接池组件 #

确保代码中所有Redis相关组件都来自redis.asyncio模块，彻底避免同步/异步组件混用。自定义连接池时必须使用异步版本：

from redis.asyncio import Redis, ConnectionPool

# 初始化异步连接池
pool = ConnectionPool(host="localhost", port=6379)
# 绑定连接池创建Redis实例
redis: Redis = Redis(connection_pool=pool)

2. 显式禁用认证（适配无密码Redis实例） #

虽然你的Redis未设置密码，但新版本redis-py可能在某些场景下默认触发认证逻辑。显式指定无认证参数可避免cred_provider被错误调用：

from redis.asyncio import Redis

redis: Redis = Redis(
    host="localhost",
    port=6379,
    username=None,
    password=None,
    decode_responses=True  # 可选，根据业务需求添加
)

3. 检查服务类中Redis实例的初始化 #

你提到后续在session_service_impl.py中执行set操作报错，需重点确认self._redis的初始化逻辑：

• 确保self._redis是从redis.asyncio导入的Redis实例，而非同步版本redis.Redis
• 排查是否在后续逻辑中意外覆盖或重新创建了错误的Redis实例

4. 保证连接池全局一致性（针对FastAPI项目） #

如果项目使用依赖注入，需确保全局只初始化一次异步Redis实例/连接池，避免每次请求创建新实例导致的连接池混乱。示例依赖注入写法：

from fastapi import Depends
from redis.asyncio import Redis, ConnectionPool

async def get_redis() -> Redis:
    pool = ConnectionPool(host="localhost", port=6379)
    redis = Redis(connection_pool=pool)
    try:
        yield redis
    finally:
        await redis.close()
        await pool.disconnect()

# 在路由或服务中使用
async def some_route(redis: Redis = Depends(get_redis)):
    await redis.set("key", "value")

内容的提问来源于stack exchange，提问作者loweryjk