o ˷e‡@sdZddlZddlZddlZddlZddlZddlZddlZddl Z ddl m Z ddl m Z mZmZmZmZmZddlmZddlmZddlmZddlmZdd lmZdd lmZdd lm Z erjdd l!m"Z"e#e$Z%Gd ddej&Z'Gdddej&Z(GdddZ)Gddde)Z*GdddZ+ej,GdddZ-e GdddZ.e GdddZ/e Gddde-Z0e Gdd d e+Z1e Gd!d"d"e+Z2e Gd#d$d$e+Z3e Gd%d&d&e+Z4dS)'zXHelpers for efficiently writing large amounts of data to the Google Cloud Firestore API.N) dataclass)CallableDictListOptionalUnion TYPE_CHECKING) status_pb2)_helpers)BaseDocumentReference)BulkWriteBatch) RateLimiter)BatchWriteResponse) WriteResult) BaseClientc@s(eZdZdZeZeZeZdS) BulkRetryzmszGAsyncBulkWriterMixin._with_send_mode..wrapper..) _send_moderr _executorsubmit concurrentfuturesFuture set_result)r%r"r$resultfuturer#)r"r$r%rwrapperjs   z5AsyncBulkWriterMixin._with_send_mode..wrapper) functoolswraps)r#r1rr0r_with_send_modeVs z$AsyncBulkWriterMixin._with_send_modebatch operationsBulkWriterOperationcCs\t|}|j|7_||}|j|8_|jd7_|j|7_||||dS)a(Sends a batch without regard to rate limits, meaning limits must have already been checked. To that end, do not call this directly; instead, call `_send_until_queue_is_empty`. Args: batch(:class:`~google.cloud.firestore_v1.base_batch.BulkWriteBatch`) N)len_in_flight_documents_send_total_batches_sent_total_write_operations_process_response)r%r5r6 _len_batchresponserrr _send_batchxs  z AsyncBulkWriterMixin._send_batchr@returnNc Cst|j}||||t|jD]5\}}|jdkr)||||j||q||}| t ||j|j d|}|rH|j d7_ | |qdS)zInvokes submitted callbacks for each batch and each operation within each batch. As this is called from `_send_batch()`, this is parallelized if we are in that mode. r) operationcodemessager8N)list_document_referencesvalues_batch_callback enumeratestatusrD_success_callback write_results_error_callbackBulkWriteFailurerEattempts_retry_operation) r%r5r@r6batch_referencesindexrKrC should_retryrrrr>s2     z&AsyncBulkWriterMixin._process_responserCcCsbd}|jjtjkr|jd}n |jjtjkr|j}tjtj|d}t |j t ||ddS)Nr)seconds)rCrun_at) _optionsretryrrrPrdatetimeutcnow timedeltabisectinsort_retriesOperationRetry)r%rCdelayrWrrrrQs  z%AsyncBulkWriterMixin._retry_operationcCs|S)zHook for overwriting the sending of batches. As this is only called from `_send_batch()`, this is parallelized if we are in that mode. )commitr%r5rrrr;szAsyncBulkWriterMixin._send)rrrrr4r rrArr>r*r+r,rQr;rrrrr Ks0 "  * r c @seZdZUdZdZeed<  dGdddedfd d Ze d e d e d dddfddZ e de ded dddfddZe ddd ddefddZdHddZddZddZdd Zd!d"Zd#d$Zd%d&Zd'd(Zd)d*Zd+d,Zdedefd-d.Z /dIde d0ed1eddfd2d3Z  /dJde d4eej d1eddfd5d6Z! 7 /dKde d0ed8e"ee#fd1eddf d9d:Z$  /dJde d;e%d4eej d1eddf de'e edgdfddfd?d@Z(d>e'e e dgdfddfdAdBZ)d>e'ddgefddfdCdDZ*dEdFZ+dS)L BulkWritera  Accumulate and efficiently save large amounts of document write operations to the server. BulkWriter can handle large data migrations or updates, buffering records in memory and submitting them to the server in batches of 20. The submission of batches is internally parallelized with a ThreadPoolExecutor, meaning end developers do not need to manage an event loop or worry about asyncio to see parallelization speed ups (which can easily 10x throughput). Because of this, there is no companion `AsyncBulkWriter` class, as is usually seen with other utility classes. Usage: # Instantiate the BulkWriter. This works from either `Client` or # `AsyncClient`. db = firestore.Client() bulk_writer = db.bulk_writer() # Attach an optional success listener to be called once per document. bulk_writer.on_write_result( lambda reference, result, bulk_writer: print(f'Saved {reference._document_path}') ) # Queue an arbitrary amount of write operations. # Assume `my_new_records` is a list of (DocumentReference, dict,) # tuple-pairs that you supply. reference: DocumentReference data: dict for reference, data in my_new_records: bulk_writer.create(reference, data) # Block until all pooled writes are complete. bulk_writer.flush() Args: client(:class:`~google.cloud.firestore_v1.client.Client`): The client that created this BulkWriter.  batch_sizeNclientroptionsBulkWriterOptionscCst|jdkr |n||_|pt|_|jj|_|||t g|_ t g|_ d|_ g|_tj|_tj|_tj|_d|_t|jj|jjd|_d|_d|_|dS)N AsyncClientTr)initial_tokensglobal_max_tokens)typer _to_sync_copy_clientrirXmoder'_reset_operations collectionsdequer__queued_batches_is_open_pending_batch_futuresrd_default_on_successrL_default_on_batchrI_default_on_errorrNr:r initial_ops_per_secondmax_ops_per_second _rate_limiterr<r=_ensure_executor)r%rgrhrrr__init__s2      zBulkWriter.__init__r5r@ bulk_writerrBcCdSr!r)r5r@rrrrrx?zBulkWriter._default_on_batch referencer.cCrr!r)rr.rrrrrwGrzBulkWriter._default_on_successerrorrOcCs |jdkS)N)rP)rrrrrryOs zBulkWriter._default_on_errorcCsg|_g|_dSr!) _operations_operations_document_pathsr%rrrrqVs zBulkWriter._reset_operationscCs*t|dddus |jjr||_dSdS)zBReboots the executor used to send batches if it has been shutdown.r(N)getattrr( _shutdown_instantiate_executorrrrrr}ZszBulkWriter._ensure_executorcCs||dSr!)r}_send_until_queue_is_emptyrrrr_ensure_sending_s zBulkWriter._ensure_sendingcCs tjSr!)r*r+ThreadPoolExecutorrrrrrcs z BulkWriter._instantiate_executorcCsn|jjrdS |jr|q|js|jr|tdq|j r/|j }g|_ t j |q |j dS)z~ Block until all pooled write operations are complete and then resume accepting new write operations. NTg?)r(rr_enqueue_current_batchrtr_rtimesleeprvr*r+waitshutdown)r%_batchesrrrflushfs"   zBulkWriter.flushcCsd|_|dS)z| Block until all pooled write operations are complete and then reject any further write operations. FN)rurrrrrcloses zBulkWriter.closecCs t|j|jkr|dSdS)zz Checks to see whether the in-progress batch is full and, if it is, adds it to the sending queue. N)r9rrfrrrrr_maybe_enqueue_current_batchs z'BulkWriter._maybe_enqueue_current_batchcCs"|j|j||dS)zAdds the current batch to the back of the sending line, resets the list of queued ops, and begins the process of actually sending whatever batch is in the front of the line, which will often be a different batch. N)rtappendrrqrrrrrrs z!BulkWriter._enqueue_current_batchcCsv||jr9|j}|t|t|jd}|D]}||q|j||d}|j |||jsdSdS)aFirst domino in the sending codepath. This does not need to be parallelized for two reasons: 1) Putting this on a worker thread could lead to two running in parallel and thus unpredictable commit ordering or failure to adhere to rate limits. 2) This method only blocks when `self._request_send()` does not immediately return, and in that case, the BulkWriter's ramp-up / throttling logic has determined that it is attempting to exceed the maximum write speed, and so parallelizing this method would not increase performance anyway. Once `self._request_send()` returns, this method calls `self._send_batch()`, which parallelizes itself if that is our SendMode value. And once `self._send_batch()` is called (which does not block if we are sending in parallel), jumps back to the top and re-checks for any queued batches. Note that for sufficiently large data migrations, this can block the submission of additional write operations (e.g., the CRUD methods); but again, that is only if the maximum write speed is being exceeded, and thus this scenario does not actually further reduce performance. )rg)r5r6N) _schedule_ready_retriesrtpopleft _request_sendr9r ro add_to_batchrArvr)r%r6r5opr/rrrrs    z%BulkWriter._send_until_queue_is_emptycCs:t|jtj}t|D] }|j}||qdS)z+Grabs all ready retries and re-queues them.N)r]r_rZr[rangerrY)r%take_until_index_rYrrrrs    z"BulkWriter._schedule_ready_retriescCs<d} |j|jjk}|p|j|}|r|stdqdS)NFTg{Gz?)r:r|_maximum_tokens take_tokensrr)r%rfhave_received_tokensunder_thresholdrrrrs  zBulkWriter._request_sendr document_datarPcCL||j|jvr||jt|||d|j|j|dS)aAdds a `create` pb to the in-progress batch. If the in-progress batch already contains a write operation involving this document reference, the batch will be sealed and added to the commit queue, and a new batch will be created with this operation as its first entry. If this create operation results in the in-progress batch reaching full capacity, then the batch will be similarly added to the commit queue, and a new batch will be created for future operations. Args: reference (:class:`~google.cloud.firestore_v1.base_document.BaseDocumentReference`): Pointer to the document that should be created. document_data (dict): Raw data to save to the server. rrrPN)_verify_not_closed_document_pathrrrrBulkWriterCreateOperationr)r%rrrPrrrcreate  zBulkWriter.createoptioncCr)aTAdds a `delete` pb to the in-progress batch. If the in-progress batch already contains a write operation involving this document reference, the batch will be sealed and added to the commit queue, and a new batch will be created with this operation as its first entry. If this delete operation results in the in-progress batch reaching full capacity, then the batch will be similarly added to the commit queue, and a new batch will be created for future operations. Args: reference (:class:`~google.cloud.firestore_v1.base_document.BaseDocumentReference`): Pointer to the document that should be created. option (:class:`~google.cloud.firestore_v1._helpers.WriteOption`): Optional flag to modify the nature of this write. rrrPN)rrrrrrBulkWriterDeleteOperationr)r%rrrPrrrdelete?rzBulkWriter.deleteFmergecCsN||j|jvr||jt||||d|j|j|dS)aAdds a `set` pb to the in-progress batch. If the in-progress batch already contains a write operation involving this document reference, the batch will be sealed and added to the commit queue, and a new batch will be created with this operation as its first entry. If this set operation results in the in-progress batch reaching full capacity, then the batch will be similarly added to the commit queue, and a new batch will be created for future operations. Args: reference (:class:`~google.cloud.firestore_v1.base_document.BaseDocumentReference`): Pointer to the document that should be created. document_data (dict): Raw data to save to the server. merge (bool): Whether or not to completely overwrite any existing data with the supplied data. rrrrPN)rrrrrrBulkWriterSetOperationr)r%rrrrPrrrsetfs  zBulkWriter.set field_updatescCsb|jjdkr td||j|jvr||jt ||||d|j|j| dS)aAdds an `update` pb to the in-progress batch. If the in-progress batch already contains a write operation involving this document reference, the batch will be sealed and added to the commit queue, and a new batch will be created with this operation as its first entry. If this update operation results in the in-progress batch reaching full capacity, then the batch will be similarly added to the commit queue, and a new batch will be created for future operations. Args: reference (:class:`~google.cloud.firestore_v1.base_document.BaseDocumentReference`): Pointer to the document that should be created. field_updates (dict): Key paths to specific nested data that should be upated. option (:class:`~google.cloud.firestore_v1._helpers.WriteOption`): Optional flag to modify the nature of this write. ExistsOptionz5you must not pass an explicit write option to update.rrrrPN) __class__r ValueErrorrrrrrrBulkWriterUpdateOperationr)r%rrrrPrrrupdates   zBulkWriter.updatecallbackcC|ptj|_dS)zISets a callback that will be invoked once for every successful operation.N)rdrwrLr%rrrron_write_resultzBulkWriter.on_write_resultcCr)zESets a callback that will be invoked once for every successful batch.N)rdrxrIrrrron_batch_resultrzBulkWriter.on_batch_resultcCr)zYSets a callback that will be invoked once for every batch that contains an error.N)rdryrNrrrron_write_errorrzBulkWriter.on_write_errorcCs|jstddS)Nz5BulkWriter is closed and cannot accept new operations)ru ExceptionrrrrrszBulkWriter._verify_not_closed)NN)rBN)r)Nr)Fr),rrrrrfint__annotations__rr~ staticmethodr rrxr rrwboolryrqr}rrrrrrrrrrrr WriteOptionrrrFrdictrrrrrrrrrrrds  * =   81 ! * +  0 0   rdc@seZdZdZdefddZdS)r7aParent class for all operation container classes. `BulkWriterOperation` exists to house all the necessary information for a specific write task, including meta information like the current number of attempts. If a write fails, it is its wrapper `BulkWriteOperation` class that ferries it into its next retry without getting confused with other similar writes to the same document. r5cCst|tsJt|tr|j|j|jdSt|tr#|j|j|jdSt|t r3|j |j|j|j dSt|t rC|j |j|j|jdStd|jjd)z"Adds `self` to the supplied batch.)rr)rr)rrr)rrrUnexpected type of z for batch) isinstancer rrrrrrrrrrrrr TypeErrorrrrcrrrrs4    z BulkWriterOperation.add_to_batchN)rrrrr rrrrrr7s r7c@s,eZdZdZd ddZdeddfd d ZdS) BaseOperationRetryzParent class for both the @dataclass and old-style `OperationRetry` classes. Methods on this class be moved directly to `OperationRetry` when support for Python 3.6 is dropped and `dataclasses` becomes universal. otherr`cCs0t|tr |j|jkSt|tjr|j|kStS)zAllows use of `bisect` to maintain a sorted list of `OperationRetry` instances, which in turn allows us to cheaply grab all that are ready to run.)rr`rWrZNotImplemented)r%rrrr__lt__s    zBaseOperationRetry.__lt__rrBNcCst|jtr|j|jj|jj|jjddSt|jtr,|j|jj|jj |jjddSt|jt rE|j |jj|jj|jj |jjddSt|jt r^|j|jj|jj|jj |jjddStd|jjjd)zCall this after waiting any necessary time to re-add the enclosed operation to the supplied BulkWriter's internal queue.rrrrrz for OperationRetry.retryN)rrCrrrrrPrrrrrrrrrrrr)r%rrrrrYs:     zBaseOperationRetry.retry)rr`)rrrrrrdrYrrrrrs  rc@sBeZdZUdZeed<dZeed<ejZ eed<e j Z e ed<dS)riirzr{rprYN) rrrrzrrr{rrrprrrYrrrrri@s  ric@s8eZdZUeed<eed<eed<edefddZdS)rOrCrDrErBcCs|jjSr!)rCrPrrrrrPOszBulkWriteFailure.attemptsN) rrrr7rrstrpropertyrPrrrrrOHs rOc@s$eZdZUdZeed<ejed<dS)r`zRContainer for an additional attempt at an operation, scheduled for the future.rCrWN)rrrrr7rrZrrrrr`Ts r`c@s.eZdZUdZeed<eed<dZeed<dS)rz-Container for BulkWriter.create() operations.rrrrPN) rrrrr rrrPrrrrrr]s rc@s<eZdZUdZeed<eed<eej ed<dZ e ed<dS)rz-Container for BulkWriter.update() operations.rrrrrPN) rrrrr rrrr rrPrrrrrrfs rc@sBeZdZUdZeed<eed<dZee e fed<dZ e ed<dS) rz*Container for BulkWriter.set() operations.rrFrrrPN) rrrrr rrrrrrFrPrrrrrrps rc@s4eZdZUdZeed<eejed<dZ e ed<dS)rz-Container for BulkWriter.delete() operations.rrrrPN) rrrrr rrr rrPrrrrrrzs r)5rr]rrconcurrent.futuresr*rZrr2loggingr dataclassesrtypingrrrrrr google.rpcr google.cloud.firestore_v1r 'google.cloud.firestore_v1.base_documentr $google.cloud.firestore_v1.bulk_batchr &google.cloud.firestore_v1.rate_limiterr )google.cloud.firestore_v1.types.firestorer%google.cloud.firestore_v1.types.writer%google.cloud.firestore_v1.base_clientr getLoggerrloggerEnumrrr rdr7total_orderingrrirOr`rrrrrrrrs^             +8