+\h!sdZddlmZddlZddlZddlZddlmZddlm Z ddl m Z m Z m Z mZmZmZmZddlmZddlmZdd lmZmZdd lmZmZdd lmZdd lmZm Z m!Z!m"Z"m#Z#dd l$m%Z%m&Z&m'Z'ddl(m)Z)m*Z*m+Z+m,Z,ddl-m.Z.ddl/m0Z0m1Z1m2Z2ddl3m4Z4m5Z5m6Z6m7Z7m8Z8m9Z9m:Z:m;Z;ddlm?Z?e rddl@mAZAddlBmCZCddlDmEZEddlFmGZGmHZHmIZIdZJGddZKy)z> Y>> $D - - %D $ $rTctd|t|tsd|vs t|d<|jj t |fy)z*Add an insert document to the list of ops.document_idN)r isinstancerr rBappendr%)rPr]s rR add_insertz_AsyncBulk.add_inserts=!*h78_5(9J&jHUO (+,rTc t||||d} ||| d<| d|_|| d<| d|_|| d<| d|_|| d<| d|_|| d<|rd |_|j jt| fy) z8Create an update document and add it to the list of ops.)qumultiNupsertT collation arrayFiltershintsortF) rrErFrGrIrJrBr`r&) rPselectorupdatererfrg array_filtersrirjcmds rR add_updatez_AsyncBulk.add_updates v&$,6EJ  "CM  "&D (C   $&*D #"/C   $(D !CK  !DNCK  %D  #'rTct|||d}|||d<| d|_||d<| d|_||d<| d|_||d<|jj t |fy)z8Create a replace document and add it to the list of ops.)rcrdNrfTrgrirj)rrErGrIrBr`r&)rPrk replacementrfrgrirjrns rR add_replacez_AsyncBulk.add_replaces~ ,$,;?  "CM  "&D (C   $(D !CK  !DNCK #'rTc||d}| d|_||d<| d|_||d<|tk(rd|_|jj t |fy)z7Create a delete document and add it to the list of ops.)rclimitNTrgriF)rErHrrJrBr`r$)rPrkrtrgrirns rR add_deletez_AsyncBulk.add_deletesa%-u=  "&D (C   $(D !CK K  %D  #'rTc#Kd}t|jD]F\}\}}| t|}n|j|k7r|t|}|j ||H|yw)ziGenerate batches of operations, batched by type of operation, in the order **provided**. N) enumeraterBrop_typeadd)rPrunidxrx operations rR gen_orderedz_AsyncBulk.gen_orderedsl)2488)< $ %C%'9{7m' 7m GGC #  $ sA&A(c#Kttttttg}t |j D]\}\}}||j |||D]}|j s|yw)zbGenerate batches of operations, batched by type of operation, in arbitrary order. N)rr%r&r$rwrBry)rP operationsr{rxr|rzs rR gen_unorderedz_AsyncBulk.gen_unorderedst7mT']DMB )2488)< 4 %C%'9 w  # #C 3 4 Cww  s A3A=6A=cK|||j<tjtjrt tt j|jj|tt||j|||jj|jj|jj d|jj d|jj" |j$r|j'||| |jj)|||j*d{}t,j,j/|j0z }tjtjrt tt j2|jj||tt||j|||jj|jj|jj d|jj d|jj"|j$r|j5||||j7||j8d{|S7]7 #t:$r} t,j,j/|j0z }t=| t>t@fr | jB} n tE| } tjtjrt tt jF|jj|| tt||j|||jj|jj|jj d|jj d|jj"t=| t@|j$r|jI|| |t=| t>t@fr/|j7| jB|j8d{7d} ~ wwxYww)zCA proxy for SocketInfo.write_command that handles event publishing.r messageclientIdcommand commandName databaseName requestId operationIddriverConnectionIdserverConnectionId serverHost serverPort serviceIdN rr durationMSreplyrrrrrrrrrrrrfailurerrrrrrrrrisServerSideError)%fieldr! isEnabledForloggingDEBUGr#r"STARTED_topology_settings _topology_idnextiterdb_nameconnidserver_connection_idaddress service_idpublish_start write_commandcodecdatetimenow start_time SUCCEEDED_succeed_process_responsesession Exceptionr_rrdetailsr(FAILED_fail) rPbwcrn request_idmsgdocsrWrdurationexcrs rRrz_AsyncBulk.write_commands,CII  ' ' 6 -5522?? cO [[$&#&88;;#&88#@#@88++A.88++A.((--  ;; JJsJ -5 ((00S#))LLE((,,.?H++GMM:#1;;#66CC' $T#Y!$( *'*xx{{'*xx'D'D"xx//2"xx//2!hh11 {{ Z9**5#++> > >@ kM* ? ((,,.?H#1ABC(+ ,S1++GMM:#188#66CC'# $T#Y!$( *'*xx{{'*xx'D'D"xx//2"xx//2!hh11&06F&G${{ *gx8#1ABC..s{{CKKHHH = s\DQ,*J%J EJ%J#J%Q, J%#J%% Q)/F.Q$Q Q$$Q))Q,cKtjtjrt tt j |jj|tt||j|||jj|jj|jjd|jjd|jj  |j"r|j%|||} |jj'||d{}t(j(j+|j,z } |t/|j0||} nddi} tjtjrt tt j2|jj| | tt||j|||jj|jj|jjd|jjd|jj |j"r|j5|| | |S7W#t6$r} t(j(j+|j,z } t9| t:r"t/|j0|| j<} n(t9| t>r | j<} n tA| } tjtjrt tt jB|jj| | tt||j|||jj|jj|jjd|jjd|jj t9| t:|j"r!|j,J|jE|| | d} ~ wwxYww)zFA proxy for AsyncConnection.unack_write that handles event publishing.rrrNokrr)#r!rrrr#r"rrrrrrrrrrrrr unack_writerrrr)namerrrr_rrrr(rr) rPrrnrr max_doc_sizerrWresultrrrrs rRrz_AsyncBulk.unack_writeIs  ' ' 6 -5522?? cO [[$&#&88;;#&88#@#@88++A.88++A.((--  ;;**S*d3C8 88//\BBF((,,.?H!-chhVDq "// >' 5 ? ?!'!:!:!G!G#+#$(cO%([[",$.+.88;;+.88+H+H#&88#3#3A#6#&88#3#3A#6"%(("5"5 {{ Z9> qC4 ((,,.?H#/0(=chhS[[(YC1++,S1++GMM:#188#66CC'# $T#Y!$( *'*xx{{'*xx'D'D"xx//2"xx//2!hh11&06F&G"{{~~111 *gx8 ; s>DP? J)J*EJ>P?J P< F*P77P<<P?c TK|jr^|j||\}}}|jj|j|t d|j |d{|S|j||\}} }|j|||| d||d{|S7=7w)Nr)w) write_concernrrW)rO batch_commandrrrr-rr) rPrrnrBrW_ batched_cmdto_sendrrs rR_execute_batch_unackz_AsyncBulk._execute_batch_unacks   &)&7&7S&A #A{G(("" *Q/  #   (+'8'8c'B $JW ""3Za&Q Q Q!  Rs$A$B(&B$'6B(B&B(&B(cZK|jr`|j||\}}}|jj|j||j |j |d{}||fS|j||\} } }|j||| | ||d{}||fS7@7 w)N)r:rrW)rOrrrrrrr) rPrrnrBrWrrrrrrs rR_execute_batchz_AsyncBulk._execute_batchs   &)&7&7S&A #A{G88++ !ii  ,Fw(+'8'8c'B $JW--c3 CRXYYFwZs$A$B+&B''7B+B) B+)B+c K|jjj} |jjj} | j} |j st ||_d|_|j } |j| |d} | rS|jst |d|_|jd} t| j}|j| |||| || j|jj}| jt| j kr| r+t| j | jz dk(r|xs|}||jjd|j"i}|j$r|j$|d<t'j(|||j*|j*|d<|j,'| jt.t0fvr|j,|d<|rG|r#|j2s|j5d|_|j7||t8j:||j=||| |j?||jA| |tC| j | jd}|jDr|jG|||| d{\}}|jIdi}|jId d tJvr8tMjN|}tQ| || j|tS|tQ| || j|d|_ d|_|j"r!d |vrn^|jU|||| d{}| xjt|z c_| jt| j kr|j"r|d ry|jx|_} | rRyy7*7tw) NFTrrAr@bypassDocumentValidationr;writeConcernErrorcoder writeErrors)+r?rVrrW_event_listenersrMrrNvalidate_sessionrKrrxr[r: idx_offsetlenrBrAr@rapply_write_concernrDr;r$r&rL_start_retryable_write _apply_tor,PRIMARYsend_cluster_timeadd_server_api apply_timeoutr acknowledgedrgetr copydeepcopyrrr)rP generatorrrrop_id retryable full_resultfinal_write_concernrrW listenersrzlast_runcmd_namerrnrBrrwcefulls rR_execute_commandz_AsyncBulk._execute_commandsS//**//))00++ #ID  DM fg.== $Y 5 ==(#H -H%% -- C..3sww</SWW!>1 D$7$H=M!5!5y$,,O<<%)\\C N))#}=&&26:6I6IC2388'CKKGW;M,M!%CJ!)E)E6687;4%%c9n6L6LdS&&sGV<##C(""63/SWWcnnd;!--,0,?,?S#v,V&VOFG!**%8"=Cwwvq)-CC $}}[9&sD#..&I/5"3 S^^VL%*DM38D0|| (?$($=$=c3V$TTG#g,.g..3sww</n|| M :%)]] 2D sY^'W(Us2J=O.?O)B7O.7O,8AO.>'O.'O.,O.c & Kggdddddgdt d fd }jjj}|j j ||| d{}dsdr t S7w) zExecute using write commands.rrwriteConcernErrors nInserted nUpsertednMatched nModifiednRemovedupsertedc PKj|||d{y7wN)r)rrrrrrrPrs rRretryable_bulkz2_AsyncBulk.execute_command..retryable_bulkNs6''  s &$&)bulk operation_idNrr)rOptional[AsyncClientSession]rr0rboolreturnNone)r+r?rVrW_retryable_writerJr) rPrrrr|rrWrrrs ``` @@rRexecute_commandz_AsyncBulk.execute_command9s"$    1 9H UY    ))00))       *   } %5I)J #K 0 sA,B3B4Bc rK|jjj}|jjj}|j}t }|j st||_|j }|r0t|j}|j|||||d|j|jj} |jt|jkr||jjdddddii} |j| t!|j|jd} |j#| | | |d{} |xjt| z c_ |jt|jkrt|dx|_}|r/yy7^w)zCExecute write commands with OP_MSG and w=0 writeConcern, unordered.NrAF writeConcernrr)r?rVrrWrr+rMrrrxr[r:rrrBrrr) rPrrrrWrrrzrrrnrBrs rRexecute_op_msg_no_resultsz$_AsyncBulk.execute_op_msg_no_resultsism//**//))00++  #ID  -H%% -- C..3sww</doo22u"S!H ##C(SWWcnnd; $ 9 9#sC PP#g,...3sww</&*)T%: :D s1,QsEF7F5AF7F73F7c Kggdddddgd}t}t} |j||d||d||d{y7#t$rYywxYww)zAExecute write commands with OP_MSG and w=0 WriteConcern, ordered.rrNF)r-r+rr)rPrrrrinitial_write_concernrs rRexecute_command_no_resultsz%_AsyncBulk.execute_command_no_resultss"$  !-  ''%       s8 AAAAAA AAAAc0K|jr td|jr td|xr |j }|r&|jr|j dkr td|r&|j r|j dkr td|r&|jr|j dkr td|jr td |jr|j|||d {S|j||d {S77w) z3Execute all operations, returning no results (w=0).z3Collation is unsupported for unacknowledged writes.z6arrayFilters is unsupported for unacknowledged writes. zPMust be connected to MongoDB 4.4+ to use hint on unacknowledged delete commands.zPMust be connected to MongoDB 4.2+ to use hint on unacknowledged update commands.zPMust be connected to MongoDB 8.0+ to use sort on unacknowledged update commands.zGCannot set bypass_document_validation with unacknowledged write concernN) rErrFrrHmax_wire_versionrGrIrDrrArr)rPrrrunacks rRexecute_no_resultsz_AsyncBulk.execute_no_resultss   $%Z[ [  " "$%]^ ^@m&@&@"@ T**t/D/Dq/H$b  T**t/D/Dq/H$b  T^^(=(=(B$b    "Y  <<88y-XX X33D)DDDYDs$C1D3D4D DDDcK|js td|jr tdd|_|xs|jj}t ||}|j r|j}n|j}|jjj}|jsP|j||d{4d{}|j|||d{ dddd{y|j||||d{S7V7O767'#1d{7swYyxYw7!w)zExecute operations.zNo operations to executez*Bulk operations can only be executed once.TN)rBrrCr?rrrAr}rrVrWr_conn_for_writesr r)rPrrr|rrW connections rRexecutez_AsyncBulk.executes)xx"#=> > =="#OP P %F)F)F 1'=I <<((*I**,I))00))#44WiHH  J--j)]SSS   --iQZ[[ [ I S    \sxC E D# ED%ED+0D'1D+6 ED)EEE%E'D+)E+D=1D4 2D=9E)NN) r?zAsyncCollection[_DocumentType]rArrQOptional[bool]r@z Optional[str]r;z Optional[Any]rr)rzType[_BulkWriteContext])r]r1rr)NNNN)rkMapping[str, Any]rlz#Union[Mapping[str, Any], _Pipeline]rerrfrrgOptional[Mapping[str, Any]]rmz!Optional[list[Mapping[str, Any]]]ri Union[str, dict[str, Any], None]rjrrr)NNN)rkrrqrrfrrgrrirrjrrr) rkrrtintrgrrirrr)rzIterator[Optional[_Run]])rzIterator[_Run])rr'rnMutableMapping[str, Any]rrrbytesrlist[Mapping[str, Any]]rWr/rdict[str, Any])rr'rnrrrrrrrrrrWr/rr) r4Union[_BulkWriteContext, _EncryptedBulkWriteContext]rnrrBrrWr/rr) rrrnrrBrrWr/rz.tuple[dict[str, Any], list[Mapping[str, Any]]]r)r Iterator[Any]rr-rrrr0rrrrrrrzOptional[WriteConcern]rr) rrrr-rrr|strrr)rr0rrrr)rr0rrrr-rr)rr-rrr|rrr)__name__ __module__ __qualname____doc__rSpropertyr[rarorrrur}rrrrrrrrrrr r rTrRr5r5Ss1"&! !"2!"!"%3 !"  !"  !" !"F%%-26;?15,0(#(4( (  ( / (9(/(*( (L2615,0(#('( ( / ( / (*( (:2615 (#((/ ( / (  ((  S S&S S  S & S!S SSjV V&V V  V  V&V!V %Vp A%  !  ! 6 A%  !  8 >7;e3 e3$e3. e3  e3  e3e3.e34e3 e3N. .$.. .  .  .`%;#%;0=%; %;N""!"$ "  "H!E!E!!E$ !E  !EF\#\.\ \ \rTr5)Lr __future__rrrrcollections.abcr itertoolsrtypingrrrr r r r bson.objectidr bson.raw_bsonrpymongorr#pymongo.asynchronous.client_sessionrrpymongo.asynchronous.helpersrpymongo.bulk_sharedrrrrrpymongo.commonrrrpymongo.errorsrrrrpymongo.helpers_sharedr pymongo.loggerr!r"r#pymongo.messager$r%r&r'r(r)r*r+pymongo.read_preferencesr,pymongo.write_concernr-pymongo.asynchronous.collectionr.!pymongo.asynchronous.mongo_clientr/pymongo.asynchronous.poolr0pymongo.typingsr1r2r3_IS_SYNCr5rrTrRr6s# *#)!c7  :MM   4.?B9FF ^ \^ \rT