# Async Helpers

Async variants of all helpers are available in `elasticsearch.helpers` and are all prefixed with `async_*`. You’ll notice that these APIs are identical to the ones in the sync Helpers documentation.

All async helpers that accept an iterator or generator also accept async iterators and async generators.

## Streaming Bulk

> async elasticsearch.helpers.async_streaming_bulk(client, actions, chunk_size=500, max_chunk_bytes=104857600, flush_after_seconds=None, raise_on_error=True, expand_action_callback=<function expand_action>, raise_on_exception=True, max_retries=0, initial_backoff=2, max_backoff=600, yield_ok=True, ignore_status=(), retry_on_status=(429, ), *args, **kwargs)

>   Streaming bulk consumes actions from the iterable passed in and yields results per action. For non-streaming usecases use `async_bulk()` which is a wrapper around streaming bulk that returns summary information about the bulk operation once the entire input is consumed and sent.

>   If you specify `max_retries` it will also retry any documents that were rejected with a `429` status code. Use `retry_on_status` to configure which status codes will be retried. To do this it will wait (by calling asyncio.sleep which will block) for `initial_backoff` seconds and then, every subsequent rejection for the same chunk, for double the time every time up to `max_backoff` seconds.

>   Parameters:

>     - client (AsyncElasticsearch) – instance of `AsyncElasticsearch` to use

>     - actions (Iterable[bytes | str | Dict[str, Any] | BulkMeta] | AsyncIterable[bytes | str | Dict[str, Any] | BulkMeta]) – iterable or async iterable containing the actions to be executed

>     - chunk_size (int) – number of docs in one chunk sent to es (default: 500)

>     - max_chunk_bytes (int) – the maximum size of the request in bytes (default: 100MB)

>     - flush_after_seconds (float | None) – time in seconds after which a chunk is written even if hasn’t reached chunk_size or max_chunk_bytes. Set to 0 to not use a timeout-based flush. (default: 0)

>     - raise_on_error (bool) – raise `BulkIndexError` containing errors (as .errors) from the execution of the last chunk when some occur. By default we raise.

>     - raise_on_exception (bool) – if `False` then don’t propagate exceptions from call to `bulk` and just report the items that failed as failed.

>     - expand_action_callback (Callable[[bytes | str | Dict[str, Any]], Tuple[Dict[str, Any], None | bytes | Dict[str, Any]]]) – callback executed on each action passed in, should return a tuple containing the action line and the data line (None if data line should be omitted).

>     - retry_on_status (int | Collection[int]) – HTTP status code that will trigger a retry. (if None is specified only status 429 will retry).

>     - max_retries (int) – maximum number of times a document will be retried when retry_on_status (defaulting to `429`) is received, set to 0 (default) for no retries

>     - initial_backoff (float) – number of seconds we should wait before the first retry. Any subsequent retries will be powers of `initial_backoff * 2**retry_number`

>     - max_backoff (float) – maximum number of seconds a retry will wait

>     - yield_ok (bool) – if set to False will skip successful documents in the output

>     - ignore_status (int | Collection[int]) – list of HTTP status code that you want to ignore

>     - client

>     - actions

>     - chunk_size

>     - max_chunk_bytes

>     - flush_after_seconds

>     - raise_on_error

>     - expand_action_callback

>     - raise_on_exception

>     - max_retries

>     - initial_backoff

>     - max_backoff

>     - yield_ok

>     - ignore_status

>     - retry_on_status

>     - args (Any)

>     - kwargs (Any)

>   Return type:

>     AsyncIterable[Tuple[bool, Dict[str, Any]]]

## Bulk

> async elasticsearch.helpers.async_bulk(client, actions, stats_only=False, ignore_status=(), *args, **kwargs)

>   Helper for the `bulk()` api that provides a more human friendly interface - it consumes an iterator of actions and sends them to elasticsearch in chunks. It returns a tuple with summary information - number of successfully executed actions and either list of errors or number of errors if `stats_only` is set to `True`. Note that by default we raise a `BulkIndexError` when we encounter an error so options like `stats_only` only+ apply when `raise_on_error` is set to `False`.

>   When errors are being collected original document data is included in the error dictionary which can lead to an extra high memory usage. If you need to process a lot of data and want to ignore/collect errors please consider using the `async_streaming_bulk()` helper which will just return the errors and not store them in memory.

>   Parameters:

>     - client (AsyncElasticsearch) – instance of `AsyncElasticsearch` to use

>     - actions (Iterable[bytes | str | Dict[str, Any]] | AsyncIterable[bytes | str | Dict[str, Any]]) – iterator containing the actions

>     - stats_only (bool) – if True only report number of successful/failed operations instead of just number of successful and a list of error responses

>     - ignore_status (int | Collection[int]) – list of HTTP status code that you want to ignore

>     - client

>     - actions

>     - stats_only

>     - ignore_status

>     - args (Any)

>     - kwargs (Any)

>   Return type:

>     Tuple[int, int | List[Any]]

>   Any additional keyword arguments will be passed to `async_streaming_bulk()` which is used to execute the operation, see `async_streaming_bulk()` for more accepted parameters.

## Scan

> async elasticsearch.helpers.async_scan(client, query=None, scroll='5m', raise_on_error=True, preserve_order=False, size=1000, request_timeout=None, clear_scroll=True, scroll_kwargs=None, **kwargs)

>   Simple abstraction on top of the `scroll()` api - a simple iterator that yields all hits as returned by underlining scroll requests.

>   By default scan does not return results in any pre-determined order. To have a standard order in the returned documents (either by score or explicit sort definition) when scrolling, use `preserve_order=True`. This may be an expensive operation and will negate the performance benefits of using `scan`.

>   Parameters:

>     - client (AsyncElasticsearch) – instance of `AsyncElasticsearch` to use

>     - query (Any | None) – body for the `search()` api

>     - scroll (str) – Specify how long a consistent view of the index should be maintained for scrolled search

>     - raise_on_error (bool) – raises an exception (`ScanError`) if an error is encountered (some shards fail to execute). By default we raise.

>     - preserve_order (bool) – don’t set the `search_type` to `scan` - this will cause the scroll to paginate with preserving the order. Note that this can be an extremely expensive operation and can easily lead to unpredictable results, use with caution.

>     - size (int) – size (per shard) of the batch send at each iteration.

>     - request_timeout (float | None) – explicit timeout for each call to `scan`

>     - clear_scroll (bool) – explicitly calls delete on the scroll id via the clear scroll API at the end of the method on completion or error, defaults to true.

>     - scroll_kwargs (MutableMapping[str, Any] | None) – additional kwargs to be passed to `scroll()`

>     - client

>     - query

>     - scroll

>     - raise_on_error

>     - preserve_order

>     - size

>     - request_timeout

>     - clear_scroll

>     - scroll_kwargs

>     - kwargs (Any)

>   Return type:

>     AsyncIterable[Dict[str, Any]]

>   Any additional keyword arguments will be passed to the initial `search()` call:

>       async_scan(
>           client,
>           query={"query": {"match": {"title": "python"}}},
>           index="orders-*"
>       )

## Reindex

> async elasticsearch.helpers.async_reindex(client, source_index, target_index, query=None, target_client=None, chunk_size=500, scroll='5m', op_type=None, scan_kwargs={}, bulk_kwargs={})

>   Reindex all documents from one index that satisfy a given query to another, potentially (if target_client is specified) on a different cluster. If you don’t specify the query you will reindex all the documents.

>   Since `2.3` a `reindex()` api is available as part of elasticsearch itself. It is recommended to use the api instead of this helper wherever possible. The helper is here mostly for backwards compatibility and for situations where more flexibility is needed.

>   Note

>   This helper doesn’t transfer mappings, just the data.

>   Parameters:

>     - client (AsyncElasticsearch) – instance of `AsyncElasticsearch` to use (for read if target_client is specified as well)

>     - source_index (str | Collection[str]) – index (or list of indices) to read documents from

>     - target_index (str) – name of the index in the target cluster to populate

>     - query (Any) – body for the `search()` api

>     - target_client (AsyncElasticsearch | None) – optional, is specified will be used for writing (thus enabling reindex between clusters)

>     - chunk_size (int) – number of docs in one chunk sent to es (default: 500)

>     - scroll (str) – Specify how long a consistent view of the index should be maintained for scrolled search

>     - op_type (str | None) – Explicit operation type. Defaults to ‘_index’. Data streams must be set to ‘create’. If not specified, will auto-detect if target_index is a data stream.

>     - scan_kwargs (MutableMapping[str, Any]) – additional kwargs to be passed to `async_scan()`

>     - bulk_kwargs (MutableMapping[str, Any]) – additional kwargs to be passed to `async_bulk()`

>     - client

>     - source_index

>     - target_index

>     - query

>     - target_client

>     - chunk_size

>     - scroll

>     - op_type

>     - scan_kwargs

>     - bulk_kwargs

>   Return type:

>     Tuple[int, int | List[Any]]