a _bD @sddlZddlZddlmZddlmZddlmZmZm Z m Z ddl m Z ddl mZed Zd d Zd"ddZddZddZddd e d dddd d f ddZd#ddZd$ddZddddiifd d!ZdS)%N)map)TransportError)_ActionChunker_process_bulk_chunk_error_process_bulk_chunk_success expand_action) ScanError)AsyncOpenSearchzopensearchpy.helperscCsRt|||d}|2z&3dHW\}}|||}|r|Vq6|}|rN|VdS)zf Split actions into chunks by number or size, serialize them into strings in the process. ) chunk_sizemax_chunk_bytes serializerN)rfeedflush)actionsr r rZchunkeractiondataretrB/tmp/pip-target-4jja1joz/lib/python/opensearchpy/_async/helpers.py_chunk_actions0s  rTrc Ost|ttfs|f}z,|jd|dg|Ri|IdH}Wn6tyv} zt| ||||d} WYd} ~ nd} ~ 00t||||d} | D] } | VqdS)zC Send a bulk request to opensearch and process the output.  N)error bulk_data ignore_statusraise_on_exceptionraise_on_error)resprrr) isinstancelisttupleZbulkjoinrrr) client bulk_actionsrrrrargskwargsregenitemrrr_process_bulk_chunkAs( ,r*cs6tdrStdr Sfdd}|S)z:Turns an async iterable or iterable into an async iterator __anext__ __aiter__csD] }|VqdSNrr)xrrfnszaiter..f)hasattrr,)r0r1rr/raitergs    r3cGsFdd|D}z tdd|DIdHVqWnty@Yn0dS)zbZips async iterables and iterables into an async iterator with the same behavior as zip() cSsg|] }t|qSrr3.0r0rrr yzazip..csg|]}|IdHqSr-)r+r5rrrr7|r8N)r!StopAsyncIteration) iterablesZaitersrrrazipus   r;ii@iXc  sfdd}t||||jj2z^3dHW\}}t|dD]>}gg}}|rvtt| |d|dIdHzt|t|||||| g| Ri| 2z3dHW\}\}}|s| \}}|r|ddkr|d|kr| t |jjj || |n|||ifVq| r||fVq6Wn>tyj}z$||ksT|jdkrVWYd}~q@d}~00|svq"||}}q@q"6dS)a  Streaming bulk consumes actions from the iterable passed in and yields results per action. For non-streaming usecases use :func:`~opensearchpy.helpers.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. To do this it will wait (**by calling asyncio.sleep**) for ``initial_backoff`` seconds and then, every subsequent rejection for the same chunk, for double the time every time up to ``max_backoff`` seconds. :arg client: instance of :class:`~opensearchpy.AsyncOpenSearch` to use :arg actions: iterable or async iterable containing the actions to be executed :arg chunk_size: number of docs in one chunk sent to client (default: 500) :arg max_chunk_bytes: the maximum size of the request in bytes (default: 100MB) :arg raise_on_error: raise ``BulkIndexError`` containing errors (as `.errors`) from the execution of the last chunk when some occur. By default we raise. :arg raise_on_exception: if ``False`` then don't propagate exceptions from call to ``bulk`` and just report the items that failed as failed. :arg expand_action_callback: 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). :arg max_retries: maximum number of times a document will be retried when ``429`` is received, set to 0 (default) for no retries on ``429`` :arg initial_backoff: number of seconds we should wait before the first retry. Any subsequent retries will be powers of ``initial_backoff * 2**retry_number`` :arg max_backoff: maximum number of seconds a retry will wait :arg yield_ok: if set to False will skip successful documents in the output :arg ignore_status: list of HTTP status code that you want to ignore cs&t2z3dHW}|Vq6dSr-r4r.rexpand_action_callbackrr map_actionssz)async_streaming_bulk..map_actionsNr rstatusi)r transportrrangeasynciosleepminr;r*popitemextendrdumpsappendr status_code)r#rr r rr=r max_retriesZinitial_backoffZ max_backoffyield_okrr%r&r>rr$attemptZto_retryZ to_retry_datarokinforr'rr<rasync_streaming_bulks^3      rOFc sxd\}}g}d|d<t||g|Rd|i|2z43dHW\} } | s\|sR|| |d7}q0|d7}q06||rr|n|fS)a Helper for the :meth:`~opensearchpy.AsyncOpenSearch.bulk` api that provides a more human friendly interface - it consumes an iterator of actions and sends them to opensearch 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 :func:`~opensearchpy.helpers.async_streaming_bulk` helper which will just return the errors and not store them in memory. :arg client: instance of :class:`~opensearchpy.AsyncOpenSearch` to use :arg actions: iterator containing the actions :arg stats_only: if `True` only report number of successful/failed operations instead of just number of successful and a list of error responses :arg ignore_status: list of HTTP status code that you want to ignore Any additional keyword arguments will be passed to :func:`~opensearchpy.helpers.async_streaming_bulk` which is used to execute the operation, see :func:`~opensearchpy.helpers.async_streaming_bulk` for more accepted parameters. )rrTrKrNr )rOrH) r#r stats_onlyrr%r&successfailederrorsrMr)rrr async_bulks$   rT5mc  Ks|pi}|s$|r|ni}d|d<i} dD]} | | vr,| | | | <q,|durp| rp| D]\} } || | qZ|jf||||d| IdH} | d}z|r`| ddr`| ddD] }|Vq| dd d }| dd d }| dd d }|||kr4d }t|||||r4t|||||f|jfd||di|IdH} | d}qW|r|r|j fdd|gii| ddiIdHn6|r|r|j fdd|gii| ddiIdH0dS)au Simple abstraction on top of the :meth:`~opensearchpy.AsyncOpenSearch.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``. :arg client: instance of :class:`~opensearchpy.AsyncOpenSearch` to use :arg query: body for the :meth:`~opensearchpy.AsyncOpenSearch.search` api :arg scroll: Specify how long a consistent view of the index should be maintained for scrolled search :arg raise_on_error: raises an exception (``ScanError``) if an error is encountered (some shards fail to execute). By default we raise. :arg preserve_order: 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. :arg size: size (per shard) of the batch send at each iteration. :arg request_timeout: explicit timeout for each call to ``scan`` :arg clear_scroll: 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. :arg scroll_kwargs: additional kwargs to be passed to :meth:`~opensearchpy.AsyncOpenSearch.scroll` Any additional keyword arguments will be passed to the initial :meth:`~opensearchpy.AsyncOpenSearch.search` call:: async_scan(client, query={"query": {"match": {"title": "python"}}}, index="orders-*", doc_type="books" ) Z_docsort)headersZapi_keyZ http_authN)bodyscrollsizerequest_timeoutZ _scroll_idhitsZ_shardsZ successfulrskippedtotalzGScroll request has only succeeded on %d (+%d skipped) shards out of %d.rY) scroll_idrZr`ignore)i) copyitems setdefaultsearchgetloggerwarningr rZ clear_scroll)r#queryrZrZpreserve_orderr[r\riZ scroll_kwargsr&Ztransport_kwargskeyvalrr`hitZshards_successfulZshards_skippedZ shards_totalZshards_messagerrr async_scan%s3       rnc sd|dur |n|}t|f|||d|} dd} ddi} | |t|| | |fd|i| IdHS)a 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 :meth:`~opensearchpy.AsyncOpenSearch.reindex` api is available as part of opensearch 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. :arg client: instance of :class:`~opensearchpy.AsyncOpenSearch` to use (for read if `target_client` is specified as well) :arg source_index: index (or list of indices) to read documents from :arg target_index: name of the index in the target cluster to populate :arg query: body for the :meth:`~opensearchpy.AsyncOpenSearch.search` api :arg target_client: optional, is specified will be used for writing (thus enabling reindex between clusters) :arg chunk_size: number of docs in one chunk sent to client (default: 500) :arg scroll: Specify how long a consistent view of the index should be maintained for scrolled search :arg scan_kwargs: additional kwargs to be passed to :func:`~opensearchpy.helpers.async_scan` :arg bulk_kwargs: additional kwargs to be passed to :func:`~opensearchpy.helpers.async_bulk` N)rjindexrZcSs>|2z23dHW}||d<d|vr0||d|Vq6dS)N_indexfields)updatepop)r]rohrrr_change_doc_indexs z(async_reindex.._change_doc_indexrPTr )rnrrrT) r#Z source_indexZ target_indexrjZ target_clientr rZZ scan_kwargsZ bulk_kwargsdocsrur&rrr async_reindexs&* rw)TTr)Fr)NrUTFrVNTN)rBloggingcompatr exceptionsrZhelpers.actionsrrrrZhelpers.errorsr r#r getLoggerrgrr*r3r;rOrTrnrwrrrrsV      & q 6 |