opensearch-py 2.3.1版本AsyncOpenSearch客户端AWS签名连接失败求助

错误原因 #

你用了同步的RequestsHttpConnection连接类搭配异步的AsyncOpenSearch客户端，导致核心冲突：同步连接的perform_request方法直接返回tuple，而异步代码中用await等待非协程对象，触发TypeError。同时，requests_aws4auth是同步签名库，也不适配异步场景。

解决方案 #

1. 替换为异步连接类

用异步客户端专属的AIOHttpConnection替代RequestsHttpConnection，这是异步场景的标准连接类。

2. 实现异步AWS签名

使用aiobotocore（异步版boto3）生成AWS签名，替代同步的requests_aws4auth。

完整代码示例 #

先安装依赖：

pip install opensearchpy aiobotocore

代码：

import asyncio
from opensearchpy import AsyncOpenSearch, AIOHttpConnection
from aiobotocore.session import get_session

class AsyncAWS4Auth:
    def __init__(self, aws_access_key=None, aws_secret_key=None, region=None, service="es"):
        self.aws_access_key = aws_access_key
        self.aws_secret_key = aws_secret_key
        self.region = region
        self.service = service
        self.session = get_session()

    async def __call__(self, request):
        # 构造完整请求URL
        request_url = f"{request.url.scheme}://{request.url.netloc}{request.url.path}"
        if request.url.query:
            request_url += f"?{request.url.query}"
        
        # 创建异步客户端并生成签名
        async with self.session.create_client(
            self.service,
            region_name=self.region,
            aws_access_key_id=self.aws_access_key,
            aws_secret_access_key=self.aws_secret_key
        ) as client:
            signer = client._request_signer
            signed_request = signer.sign(
                request.method,
                request_url,
                request.headers,
                request.body,
                expires_in=3600
            )
            # 更新请求头为签名后的内容
            request.headers.update(signed_request.headers)
            return request

async def main():
    # 替换为你的OpenSearch配置
    host = "your-opensearch-endpoint"
    aws_access_key = "your-access-key"
    aws_secret_key = "your-secret-key"
    aws_region = "your-region"

    auth = AsyncAWS4Auth(aws_access_key, aws_secret_key, aws_region, "es")
    es_client = AsyncOpenSearch(
        hosts=[{"host": host, "port": 443}],
        http_auth=auth,
        connection_class=AIOHttpConnection,
        use_ssl=True,
        verify_certs=True,
        timeout=120
    )

    # 测试连接
    response = await es_client.info()
    print(response)

if __name__ == "__main__":
    asyncio.run(main())

额外说明 #

• 如果在AWS托管环境（EC2、ECS、Lambda）运行，可以省略aws_access_key和aws_secret_key，aiobotocore会自动获取IAM角色的临时凭证。
• 确保service参数正确：AWS OpenSearch服务的签名服务名是es，旧版Elasticsearch服务也可使用该值。

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