o Lj,@s~dZddlmZmZddlmZmZmZmZm Z m Z m Z m Z ddZ dddZdd d Zd d Zd dZddZddZdS)a[Shared HIMPORT wire-execution helpers for the asynchronous clients. Async mirror of :mod:`redis._himport_exec`. The PREPARE / SET / DISCARD packed-write drain loops, per-connection version bookkeeping, and ``NoSuchFieldsetError`` re-prepare-and-retry are identical for the standalone (:class:`redis.asyncio.Redis`) and cluster (async ``ClusterNode``) clients, differing only in (a) which object provides ``parse_response`` and (b) the cluster-only ASK-redirect handling. These coroutines take that object as ``node`` and an ``asking`` flag (``False`` -- and a no-op -- for standalone), so both clients share one implementation. The per-class ``_himport_*`` methods are thin delegators to these coroutines. The sync version lives in :mod:`redis._himport_exec`; the two are kept separate on purpose (the project maintains parallel sync/async stacks by hand). )NoSuchFieldsetError ResponseError)HIMPORT_DISCARDHIMPORT_PREPARE HIMPORT_SETHImportRegistryhimport_discard_commandhimport_prepare_commandhimport_set_commandparse_himport_set_argsc s|j}|dus|j|jkrdS|j}|t|j}|rc||dd|DIdHd}|D]+}z ||t IdHWnt yT}z |pI|}WYd}~nd}~ww|j |dq1|durc|||_dS)akDISCARD, on ``conn``, any prepared fieldset removed from the registry. Runs at most once per registry mutation: the connection records the registry ``revision`` it last reconciled against, so unchanged registries are a no-op. ``node`` supplies ``parse_response`` (the standalone client itself, or the owning ``ClusterNode`` in cluster mode). NcSsg|]}t|qS)r).0nr r Y/home/djax/ivt_ai_plugin/venv/lib/python3.10/site-packages/redis/asyncio/_himport_exec.py 4sz&reconcile_discards..) himport_registry_himport_reconciled_revisionrevisionnames_to_discardlist_himport_preparedsend_packed_command pack_commandsparse_responserrpop)nodeconnregistry reconciled_tostale first_errorrer r rreconcile_discardss,  r"Fc sBt||jg}|r|d|t||||||IdHd}} } d} z ||tIdHWntyJ} z| }WYd} ~ nd} ~ ww|rmz ||dIdHWntyl} z| } WYd} ~ nd} ~ wwz ||t IdH} Wnty} z| } WYd} ~ nd} ~ ww|r||j |j |<| r| | r| | S)aPREPARE ``fieldset`` bundled with the SET on ``conn`` (one packed write). When ``asking`` is set (an ASK-redirected cluster SET) the batch becomes ``[PREPARE, ASKING, SET]`` so the per-command ASKING allowance falls immediately before the SET -- the only slot-scoped command. PREPARE is a connection-session command the ASKING flag does not gate, so placing it before ASKING is safe. Every reply is drained even on a per-command error so the packed replies never desync the pooled socket. ASKINGNr$) r fieldsappendr rrrrrrversionr) rrkey fieldset_namevaluesfieldsetaskingcommands prep_error ask_error set_errorset_respr!r r rprepare_and_setFsF    r2c s\t||IdH|j}|dur||nd}|dur2|j||jkr2t|||||||dIdHS|rt||dt|||gIdHz | |dIdHWn/t ys}zz | |t IdHW|t ynY|wd}~ww|j t|||IdHz | |t IdHWSt y|dur|j|dt|||||||dIdHYSw)aExecute an ``HIMPORT SET`` on ``conn`` with the required session setup. Reconciles deferred discards, lazily bundles PREPARE with the SET on first use of a fieldset, and recovers once from a mid-connection fieldset loss (``NoSuchFieldsetError``) by re-PREPARE-and-retry. When ``asking`` is set the ASKING allowance is folded into the SET's own packed write so it immediately precedes the (slot-scoped) SET; the session setup runs first, since those are connection-session commands the flag does not gate. N)r,r#r$)r"rgetrr'r2rrr rrr send_commandrr) rrr(r)r*r,rr+r/r r r execute_setssL      r5csXt|||IdH}|sdS||t|IdHt|||IdH}|dur*|dS)aqPre-flight ``conn`` for a pipeline batch containing ``HIMPORT SET``s. The packed pipeline write bypasses the per-command lazy-PREPARE path, so the fieldsets referenced by the buffered SETs must be PREPAREd on ``conn`` first. Reconciles deferred discards, then PREPAREs every distinct registered fieldset the batch references that this connection has not already prepared, in one packed write. ``command_arg_lists`` is the batch's per-command positional-arg sequences (the caller extracts them from its own command representation). No-op when the batch has no registry-backed ``HIMPORT SET``. N)pipeline_preparesrrprepare_wire_commandsdrain_pipeline_prepares)rrcommand_arg_lists to_preparer r r rprepare_pipelines  r;c st|dd}t|tsgSt||IdHg}t}|D]0}t|}|dur(q|d}||vr1q||||} | durM|j|| j krM| | q|S)aReturn the fieldsets that must be PREPAREd on ``conn`` for this batch. Like :func:`prepare_pipeline`, but does **not** send the PREPAREs: the caller folds them into the same packed write as the queued commands (see the pipeline executors), so the first pipeline use of a fieldset on a fresh or reconnected connection stays a single round trip instead of a separate PREPARE exchange followed by the batch. Deferred-discard reconciliation is still performed here, but it only touches the socket when discards are actually pending (rare); the common warm-up cost -- the first-use PREPARE -- is what gets folded. Returns an empty list when the batch references no not-yet-prepared registered fieldset, or when ``conn`` carries no real HIMPORT registry. rN) getattr isinstancerr"setr addr3rr'r&) rrr9rr:seenargsparsedr)r+r r rr6s*    r6cCsdd|DS)zLThe leading ``HIMPORT PREPARE`` wire commands the caller folds into a batch.cSsg|] }t|j|jqSr )r namer%)r fsr r rrsz)prepare_wire_commands..r ) fieldsetsr r rr7sr7c sfd}|D]+}z ||tIdHWnty(}z |p|}WYd}~qd}~ww|j|j|j<q|S)arDrain the ``len(fieldsets)`` leading PREPARE replies of a folded pipeline write, marking each fieldset prepared on success. Returns the first ``ResponseError`` (or ``None``). The caller must still drain the queued command replies and only then surface this error: every reply on the wire has to be read before raising, or the pooled socket desyncs. N)rrrr'rrD)rrrFr rEr!r r rr8s r8N)F)__doc__redis.exceptionsrr redis.himportrrrrrr r r r"r2r5r;r6r7r8r r r rs( ) -<$