U ڲgi$@sVUdZddlmZddlZddlmZddlmZmZddl m Z ddl m Z dd l mZmZdd lmZmZmZdd lmZdd lmZmZmZdd lmZerddlmZddlmZddl m!Z!m"Z"ddl#m$Z$ddl%m&Z&ddl'm(Z(dZ)de*d<e j+j,Z,e j+j-Z-e j.j/Z/e j0j1Z1e2dZ3GdddZ4Gddde4Z5Gddde4Z6dS)z commands pipeline management ) annotationsN) TracebackType)Any TYPE_CHECKING)pq)errors)PipelineCommandPQGen)DequeSelf TypeAlias)connection_summary)pipeline_communicate fetch_manysend) capabilities)PGresult) Connection)KeyPrepare) BaseCursor)BaseConnection)AsyncConnectionzEtuple[BaseCursor[Any, Any], tuple[Key, Prepare, bytes] | None] | Noner PendingResultpsycopgc@seZdZUded<ded<ddddd Zd d d d Zedd ddZedd ddZ dd ddZ dddddZ dd ddZ dd ddZ dd dd Zddd!d"d#Zd$d%dd&d'd(Zdd d)d*Zd+S), BasePipelinezDeque[PipelineCommand] command_queuezDeque[PendingResult] result_queuezBaseConnection[Any]NoneconnreturncCs0||_|j|_tt|_tt|_d|_dS)Nr)_connpgconnr r rrrlevelselfr!r(5/tmp/pip-unpacked-wheel-b_ea6rx_/psycopg/_pipeline.py__init__/s   zBasePipeline.__init__strr"cCs@|jjd|jj}t|jj}d|d|dt|ddS)N.< z at 0xx>) __class__ __module__ __qualname__rr#r$id)r'clsinfor(r(r)__repr__6s zBasePipeline.__repr__zpq.PipelineStatuscCst|jjSN)rZPipelineStatusr$Zpipeline_statusr'r(r(r)status;szBasePipeline.statusboolcCstS)zCReturn `!True` if the psycopg libpq wrapper supports pipeline mode.)r has_pipeline)r6r(r(r) is_supported?szBasePipeline.is_supportedz PQGen[None]ccsTtjdd|jdkr"|jn |js4|jjtkrB|EdH|jd7_dS)NT)checkrr) rr=r%r$Zenter_pipeline_moderZtransaction_statusACTIVE _sync_genr:r(r(r) _enter_genDs    zBasePipeline._enter_genBaseException | None)excr"c Csz|jd8_|jdkrv|jjtkrvz|jWnBtjk rt}z"|rZtd||n | dW5d}~XYnXdS)Nrrzerror ignored exiting %r: %s) r%r$r;BADZexit_pipeline_modeeZOperationalErrorloggerwarningwith_traceback)r'rDexc2r(r(r)_exitRszBasePipeline._exitccs,||EdH|jddEdHdS)NFflush) _enqueue_sync_communicate_gen _fetch_genr:r(r(r)rA`szBasePipeline._sync_genc cs4z||EdHW5|jddEdHXdS)z_ Exit current pipeline by sending a Sync and fetch back all remaining results. TrLN)rPrNrOr:r(r(r) _exit_geneszBasePipeline._exit_genc cs~t|j|jEdH}d}|D]P}|j}z|||Wqtjk rj}z|dkrZ|}W5d}~XYqXq|dk rz|dS)zqCommunicate with pipeline to send commands and possibly fetch results, which are then processed. N)rr$rrpopleft_process_resultsrFError)r'Zfetched exceptionresultsqueuedrDr(r(r)rOos zBasePipeline._communicate_gen)rMr"c cs|js dS|r(|jt|jEdHd}|jrt|jEdH}|sHq|j}z|||Wq,tjk r}z|dkr|}W5d}~XYq,Xq,|dk r|dS)a^Fetch available results from the connection and process them with pipeline queued items. If 'flush' is True, a PQsendFlushRequest() is issued in order to make sure results can be fetched. Otherwise, the caller may emit a PQpipelineSync() call to ensure the output buffer gets flushed before fetching. N) rr$Zsend_flush_requestrrrRrSrFrT)r'rMrUrVrWrDr(r(r)rPs$   zBasePipeline._fetch_genrzlist[PGresult])rWrVr"c Cs|dkrB|\}|jtkr,tj||jjdq|jtkrtdn>|\}}|rl|\}}}|jj ||||| || |dS)arProcess a results set fetched from the current pipeline. This matches 'results' with its respective element in the pipeline queue. For commands (None value in the pipeline queue), results are checked directly. For prepare statement creation requests, update the cache. Otherwise, results are attached to their respective cursor. N)encodingzpipeline aborted) r; FATAL_ERRORrFZerror_from_resultr$ _encodingPIPELINE_ABORTEDZPipelineAbortedr#Z _preparedvalidateZ_check_resultsZ _set_results) r'rWrVresultcursorZprepinfokeyprepnamer(r(r)rSs     zBasePipeline._process_resultscCs |j|jj|jddS)z#Enqueue a PQpipelineSync() command.N)rappendr$Z pipeline_syncrr:r(r(r)rNszBasePipeline._enqueue_syncN)__name__r3r4__annotations__r*r8propertyr; classmethodr>rBrKrArQrOrPrSrNr(r(r(r)r+s   rcsfeZdZUdZdZded<dddfdd Zdd d d Zd d d dZdddddddZZ S)Pipelinez(Handler for connection in pipeline mode.rzConnection[Any]r#rr cst|dSr9superr*r&r2r(r)r*szPipeline.__init__r,c Cs\z(|jj|j|W5QRXWn.tjk rV}z|dW5d}~XYnXdS)zkSync the pipeline, send any pending command and receive and process all available results. Nr#lockwaitrArFZ _NO_TRACEBACKrIr'exr(r(r)syncs  z Pipeline.syncr c Cs(|jj|j|W5QRX|Sr9r#rlrmrBr:r(r(r) __enter__s zPipeline.__enter__type[BaseException] | NonerCTracebackType | Noneexc_typeexc_valexc_tbr"c Csznz(|jj|j|W5QRXWn@tk rj}z"|rPtd||n |dW5d}~XYnXW5||XdSNz error ignored terminating %r: %s rKr#rlrmrQ ExceptionrGrHrIr'rvrwrxrJr(r(r)__exit__s  zPipeline.__exit__) rcr3r4__doc__rdr*rprrr} __classcell__r(r(rjr)rgs  rgcsfeZdZUdZdZded<dddfdd Zdd d d Zd d d dZdddddddZZ S) AsyncPipelinez.Handler for async connection in pipeline mode.rzAsyncConnection[Any]r#rr cst|dSr9rhr&rjr(r)r*szAsyncPipeline.__init__r,c spz<|jj4IdH|j|IdHW5QIdHRXWn.tjk rj}z|dW5d}~XYnXdSr9rkrnr(r(r)rps *zAsyncPipeline.syncr c s<|jj4IdH|j|IdHW5QIdHRX|Sr9rqr:r(r(r) __aenter__s&zAsyncPipeline.__aenter__rsrCrtruc szz<|jj4IdH|j|IdHW5QIdHRXWn@tk r~}z"|rdtd||n |dW5d}~XYnXW5||XdSryrzr|r(r(r) __aexit__s* zAsyncPipeline.__aexit__) rcr3r4r~rdr*rprrrr(r(rjr)rs r)7r~ __future__rloggingtypesrtypingrrrrrFabcr r _compatr r r Zpq.miscr generatorsrrrZ _capabilitiesrZpq.abcr connectionrZ _preparingrrZ _cursor_baserZ_connection_baserZconnection_asyncrrrdZ ExecStatusrYr[Z ConnStatusrEZTransactionStatusr@ getLoggerrGrrgrr(r(r(r)s:             +