Trouble using AsyncElasticsearch with RequestsHttpConnection

[sethmlarson]

(This issue is opened on behalf of @aockel and is quoted directly)

Hi, I've tried to run the simple async bulk request example and got the following error message:

File "sample_async.py", line 28, in
loop.run_until_complete(main())
File "/usr/local/Cellar/python/3.7.3/Frameworks/Python.framework/Versions/3.7/lib/python3.7/asyncio/base_events.py", line 584, in run_until_complete
return future.result()
File "sample_async.py", line 25, in main
await async_bulk(es, gendata())
File "/usr/local/lib/python3.7/site-packages/elasticsearch/_async/helpers.py", line 241, in async_bulk
async for ok, item in async_streaming_bulk(client, actions, *args, **kwargs):
File "/usr/local/lib/python3.7/site-packages/elasticsearch/_async/helpers.py", line 171, in async_streaming_bulk
**kwargs
File "/usr/local/lib/python3.7/site-packages/elasticsearch/_async/helpers.py", line 92, in azip
yield tuple([await x.anext() for x in aiters])
File "/usr/local/lib/python3.7/site-packages/elasticsearch/_async/helpers.py", line 92, in
yield tuple([await x.anext() for x in aiters])
File "/usr/local/lib/python3.7/site-packages/elasticsearch/_async/helpers.py", line 55, in _process_bulk_chunk
resp = await client.bulk("\n".join(bulk_actions) + "\n", *args, **kwargs)
File "/usr/local/lib/python3.7/site-packages/elasticsearch/_async/client/init.py", line 435, in bulk
body=body,
File "/usr/local/lib/python3.7/site-packages/elasticsearch/_async/transport.py", line 286, in perform_request
timeout=timeout,
TypeError: object tuple can't be used in 'await' expression

using the following code snipped:

import asyncio
from aws_requests_auth.boto_utils import BotoAWSRequestsAuth
from elasticsearch import AsyncElasticsearch, RequestsHttpConnection
from elasticsearch.helpers import async_bulk

es_host = 'exampleurldoesnotexist.eu-west-1.es.amazonaws.com'
auth = BotoAWSRequestsAuth(aws_host=es_host,
aws_region='eu-west-1',
aws_service='es')
es = AsyncElasticsearch(host=es_host,
port=80,
connection_class=RequestsHttpConnection,
http_auth=auth,
timeout=60)

async def gendata():
mywords = ['foo', 'bar', 'baz']
for word in mywords:
yield {
"_index": "mywords",
"doc": {"word": word},
}

async def main():
await async_bulk(es, gendata())

loop = asyncio.get_event_loop()
loop.run_until_complete(main())

[sethmlarson]

So the problem here is you're passing a connection_class which isn't capable of making asynchronous HTTP requests. RequestsHttpConnection can only make sync requests. If you are relying on authentication that only works with RequestsHttpConnection out of the box you should continue using the sync elasticsearch.Elasticsearch instance.

Does that make sense? Will close once you confirm.

[Ashish20]

Hello,

I am facing the exact same issue while trying to use delete_by_query API with the AsyncElasticSearch Client. After reading the thread, I was not clear on what the solution was. Could you please elaborate on how to solve this error?

Any help will be much appreciated!

[sethmlarson]

@Ashish20 The RequestsHttpConnection class is not compatible with the AsyncElasticsearch client, the async client at this time can only use AIOHttpConnection.

If you have to use the RequestsHttpConnection due to specific libraries being used that can't be set via configuring the AIOHttpConnection then I recommend not using the async Elasticsearch client until your use-case allows it.

[mbchoa]

Apologies for adding to this, but for anyone looking for a solution in the future. I was able to get this working courtesy of this user's SO post here: https://stackoverflow.com/questions/67079777/asyncelasticsearch-client-not-accepting-connection-with-aws-elasticsearch-attr

The general gist is that the AIOHttpConnection class was extended and overrode the perform_request co-routine to add additional AWS authentication headers to outgoing requests.