support async bulk api #90
Conversation
Codecov Report
@@ Coverage Diff @@
## master #90 +/- ##
========================================
+ Coverage 97.5% 97.7% +0.2%
========================================
Files 7 7
Lines 440 522 +82
Branches 78 94 +16
========================================
+ Hits 429 510 +81
Misses 7 7
- Partials 4 5 +1
Continue to review full report at Codecov.
|
|
Thanks for the contribution! Can You add some tests? |
|
This project still support Python 3.5.3-, can You pass everywhere explicit loop? |
aioelasticsearch/helpers.py
Outdated
| @@ -147,3 +150,121 @@ def _update_state(self, resp): | |||
| self._successful_shards = resp['_shards']['successful'] | |||
| self._total_shards = resp['_shards']['total'] | |||
| self._done = not self._hits or self._scroll_id is None | |||
|
|
|||
|
|
|||
| async def worker_bulk(client, datas: List[dict], actions: List[str], **kwargs): | |||
There was a problem hiding this comment.
All project do not provide yet typings everywhere, so it not makes to put them here. Later typings will be provided everywhere...
aioelasticsearch/helpers.py
Outdated
| async def _retry_handler(client, futures, max_retries, initial_backoff, | ||
| max_backoff, **kwargs): | ||
| finish = 0 | ||
| for attempt in range(max_retries + 1): |
There was a problem hiding this comment.
have You copied this implementation from original library?
There was a problem hiding this comment.
yes, i ported the code from https://github.com/elastic/elasticsearch-py/blob/master/elasticsearch/helpers/__init__.py#L182
aioelasticsearch/helpers.py
Outdated
| sleep = min(max_backoff, initial_backoff * 2 ** (attempt - 1)) | ||
| await asyncio.sleep(sleep) | ||
|
|
||
| results = await asyncio.gather(*futures, |
There was a problem hiding this comment.
This gather produces unexpected load, not sure, but maybe it should be rewritten in asyncio.queue / workers
aioelasticsearch/helpers.py
Outdated
| for attempt in range(max_retries + 1): | ||
| if attempt: | ||
| sleep = min(max_backoff, initial_backoff * 2 ** (attempt - 1)) | ||
| await asyncio.sleep(sleep) |
|
Ok, I am refactoring |
| from aioelasticsearch import Elasticsearch | ||
| from aioelasticsearch.helpers import bulk | ||
|
|
||
| def gen_data(): |
There was a problem hiding this comment.
What about to make it async gen? It Should be really usefull
|
I stumbled upon this because I need bulk in a asyncio project. What is the current status of this PR? Is there a particular reason why it was never merged? |
|
^^ Bulk is a really critical part of working with ES, any estimate for this? |
This comment has been minimized.
This comment has been minimized.
feature async bulk api