o ,= jYe@sddlmZddlZddlZddlZddlZddlmZddlZddl m Z ddl m Z m Z ddlmZddlmZmZmZejZeeZdZid dd d d d ddddddddddddddddddd d!d"d#d$d%d&d'd(d)d*d+iZd,Ze je je je je je j e j!e j"fZ#e jfZ$e%d-d.d/gZ&Gd0d1d1e'Z(Gd2d3d3eZ)Gd4d5d5e'Z*Gd6d7d7e'Z+d8d9Z,d:d;Z-dd?Z/Gd@dAdAe'Z0dS)B) annotationsN)Enum) exceptions)BackgroundConsumerResumableBidiRpc)_helpers) ListenRequestTarget TargetChangeiyPOK CANCELLEDUNKNOWNZINVALID_ARGUMENTZDEADLINE_EXCEEDED NOT_FOUNDZALREADY_EXISTSZPERMISSION_DENIEDZUNAUTHENTICATEDZRESOURCE_EXHAUSTEDZFAILED_PRECONDITION ZABORTED Z OUT_OF_RANGE Z UNIMPLEMENTED ZINTERNAL UNAVAILABLEZ DATA_LOSSZ DO_NOT_USEzThread-OnRpcTerminated DocTreeEntryvalueindexc@sTeZdZddZddZddZddZd d Zd d Zd dZ ddZ ddZ dS) WatchDocTreecCsi|_d|_dS)Nr)_dict_indexselfr)iC:\Users\psant\Downloads\Anu Raj Backend\school\venv\Lib\site-packages\google/cloud/firestore_v1/watch.py__init__Os zWatchDocTree.__init__cCst|jSN)listr%keysr'r)r)r*r.SszWatchDocTree.keyscCs"t}|j|_|j|_|}|Sr,)r$r%copyr&)r(Zwdtr)r)r*_copyVs  zWatchDocTree._copycCs,|}t||j|j|<|jd7_|S)Nr )r0r!r&r%)r(keyr"r)r)r*insert]szWatchDocTree.insertcCs |j|Sr,r%r(r1r)r)r*findc zWatchDocTree.findcCs|}|j|=|Sr,)r0r%r4r)r)r*removefszWatchDocTree.removeccs|jD]}|VqdSr,r3r(kr)r)r*__iter__ks zWatchDocTree.__iter__cCs t|jSr,)lenr%r'r)r)r*__len__or6zWatchDocTree.__len__cCs ||jvSr,r3r8r)r)r* __contains__rr6zWatchDocTree.__contains__N) __name__ __module__ __qualname__r+r.r0r2r5r7r:r<r=r)r)r)r*r$Ks r$c@seZdZdZdZdZdS) ChangeTyper rrN)r>r?r@ADDEDREMOVEDMODIFIEDr)r)r)r*rAvsrAc@eZdZddZdS)DocumentChangecCs||_||_||_||_dS)zDocumentChange Args: type (ChangeType): document (document.DocumentSnapshot): old_index (int): new_index (int): N)typedocument old_index new_index)r(rGrHrIrJr)r)r*r+}s  zDocumentChange.__init__Nr>r?r@r+r)r)r)r*rF| rFc@rE) WatchResultcCs||_||_||_dSr,)snapshotname change_type)r(rNrOrPr)r)r*r+s zWatchResult.__init__NrKr)r)r)r*rMrLrMcCst|tjr t|S|S)z(Wraps a gRPC exception class, if needed.N) isinstancegrpcZRpcErrorrZfrom_grpc_error) exceptionr)r)r*_maybe_wrap_exceptions  rTcCs||ksJddS)Nz+Document watches only support one document.rr))Zdoc1Zdoc2r)r)r*document_watch_comparatorsrUcCt|}t|tSr,)rTrQ_RECOVERABLE_STREAM_EXCEPTIONSrSwrappedr)r)r*_should_recover rZcCrVr,)rTrQ_TERMINATING_STREAM_EXCEPTIONSrXr)r)r*_should_terminater[r]c @seZdZddZddZeddZeddZd d Zd d Z e d dZ d.ddZ ddZ ddZddZddZddZddZddZejeejeejeejeejeiZd d!Zd"d#Zd$d%Zed&d'Zd(d)Z d*d+Z!d,d-Z"dS)/WatchcCsz||_||_||_||_||_||_|j|_t |_ d|_ | |j d|_t|_i|_i|_d|_d|_|dS)a Args: firestore: target: comparator: snapshot_callback: Callback method to process snapshots. Args: docs (List(DocumentSnapshot)): A callback that returns the ordered list of documents stored in this snapshot. changes (List(str)): A callback that returns the list of changed documents since the last snapshot delivered for this watch. read_time (string): The ISO 8601 time at which this snapshot was obtained. document_snapshot_cls: factory for instances of DocumentSnapshot FN)Z_document_reference _firestore_targets _comparator_document_snapshot_cls_snapshot_callbackZ_firestore_api_api threadingLock_closing_closed_set_documents_pfx_database_string resume_tokenr$doc_treedoc_map change_mapcurrent has_pushed _init_stream)r(Zdocument_referenceZ firestoretargetZ comparatorsnapshot_callbackdocument_snapshot_clsr)r)r*r+s"   zWatch.__init__cCsP|j}t|jjjtt||jjd|_ |j |j t |j |j |_|jdS)N)Z start_rpcZshould_recoverZshould_terminateZinitial_requestmetadata)_get_rpc_requestrrd _transportlistenrZr]r_Z _rpc_metadata_rpcadd_done_callback _on_rpc_doner on_snapshot _consumerstart)r(Z rpc_requestr)r)r*rqszWatch._init_streamcCs"|||jd|jgitdt||S)a Creates a watch snapshot listener for a document. snapshot_callback receives a DocumentChange object, but may also start to get targetChange and such soon Args: document_ref: Reference to Document snapshot_callback: callback to be called on snapshot document_snapshot_cls: class to make snapshots with reference_class_instance: class make references documents)r target_idN)_client_document_pathWATCH_TARGET_IDrU)cls document_refrsrtr)r)r* for_documents zWatch.for_documentcCs>|j\}}tj||d}|||j|jtd|j||S)N)parentZstructured_query)queryr) _parentZ _parent_infor Z QueryTargetZ _to_protobufr_pbrra)rrrsrt parent_path_Z query_targetr)r)r* for_querys zWatch.for_querycCs8|jdur |j|jd<n|jddt|jj|jdS)Nrk)ZdatabaseZ add_target)rkr`poprr_rjr'r)r)r*rv+s  zWatch._get_rpc_requestcCs|d|_t|j|_dS)Nz /documents/)_documents_pfxr;_documents_pfx_len)r(Zdatabase_stringr)r)r*ri5s zWatch._set_documents_pfxcCs|jduo|jjS)zbool: True if this manager is actively streaming. Note that ``False`` does not indicate this is complete shut down, just that it stopped getting new messages. N)r} is_activer'r)r)r*r9szWatch.is_activeNcCs|jI|jr WddS|jr$|jr td|jd|j_d|_d|_|j r:|j d|j _ g|j _ d|_ d|_tdWdn1sOwY|rhtd|t |trd|t|dS)aStop consuming messages and shutdown all helper threads. This method is idempotent. Additional calls will have no effect. Args: reason (Any): The reason to close this. If None, this is considered an "intentional" shutdown. NzStopping consumer.TzFinished stopping manager.zreason for closing: %s)rgrhr}r_LOGGERdebugstopZ _on_responsercrycloseZ_initial_request _callbacksrQ Exception RuntimeError)r(reasonr)r)r*rBs2      z Watch.closecCs:tdt|}tjt|jd|id}d|_|dS)a Triggered whenever the underlying RPC terminates without recovery. This is typically triggered from one of two threads: the background consumer thread (when calling ``recv()`` produces a non-recoverable error) or the grpc management thread (when cancelling the RPC). This method is *non-blocking*. It will start another thread to deal with shutting everything down. This is to prevent blocking in the background consumer and preventing it from being ``joined()``. z.RPC termination has signaled manager shutdown.r)rOrrkwargsTN) rinforTreThread_RPC_ERROR_THREAD_NAMErdaemonr~)r(futurethreadr)r)r*r{gs   zWatch._on_rpc_donecCs |dSr,)rr'r)r)r* unsubscribezs zWatch.unsubscribecCsRtd|jdupt|jdk}|r#|jr%|jr'||j|jdSdSdSdS)Nz%on_snapshot: target change: NO_CHANGEr)rr target_idsr; read_timeropushrk)r( target_changeZ no_target_idsr)r)r*$_on_snapshot_target_change_no_change}s  z*Watch._on_snapshot_target_change_no_changecCs,td|jd}|tkrtd|dS)Nzon_snapshot: target change: ADDrz&Unexpected target ID %s sent by server)rrrrr)r(rrr)r)r*_on_snapshot_target_change_adds   z$Watch._on_snapshot_target_change_addcCsJtd|jjr|jj}|jj}nd}d}d||f}t|t||)Nz"on_snapshot: target change: REMOVErzinternal errorz Error %s: %s)rrcausecodemessagerrZfrom_grpc_status)r(rrr error_messager)r)r*!_on_snapshot_target_change_removes    z'Watch._on_snapshot_target_change_removecCstd|dS)Nz!on_snapshot: target change: RESET)rr _reset_docsr(rr)r)r* _on_snapshot_target_change_resets  z&Watch._on_snapshot_target_change_resetcCstdd|_dS)Nz#on_snapshot: target change: CURRENTT)rrrorr)r)r*"_on_snapshot_target_change_currents  z(Watch._on_snapshot_target_change_currentcCs||jr ||jd}|Sr,) startswithrr)r( document_namer)r)r*_strip_document_pfxs zWatch._strip_document_pfxc Csl|jrdS|dur|dS|j}|d}|dkri|jj}td||j |}|durJd|}t d||jt |ddSz |||jWdSt yh}z td|d}~ww|d krtd t|jjv}t|jjv} |jj} |rtd t| j|j} || j} |j| } |j| | d d| j| jd }||j| j<dS| rtdtj|j| j<dSdS|dkrtd|jj}tj|j|<dS|dkrtd|j j}tj|j|<dS|dkr"td|j!j"|#kr t dt$j%t&|jd}|'|(|)|*dSdStdd|}|jt |ddS)aSProcess a response from the bi-directional gRPC stream. Collect changes and push the changes in a batch to the customer when we receive 'current' from the listen response. Args: proto(`google.cloud.firestore_v1.types.ListenResponse`): Callback method that receives a object to NZ response_typerzon_snapshot: target change: zUnknown target change type: z on_snapshot: )rzmeth(proto) exc: document_changezon_snapshot: document changez%on_snapshot: document change: CHANGEDT) referencedataexistsr create_time update_timez%on_snapshot: document change: REMOVEDdocument_deletez$on_snapshot: document change: DELETEdocument_removez$on_snapshot: document change: REMOVEfilterzon_snapshot: filter updatez%Filter mismatch -- restarting stream.)rOrrzUNKNOWN TYPE. UHOHzUnknown listen response type: )+rglockedrrZ WhichOneofrtarget_change_typerr_target_changetype_dispatchgetr ValueErrorrrrrZremoved_target_idsrHrZ decode_dictfieldsr_rrOrbrrrnrArCrrrcount _current_sizererrr~joinrrq)r(protoZpbwhichrmethrZexc2changedremovedrHrrrrNrOrr)r)r*r|s                   zWatch.on_snapshotc Cs||j|j|\}}}||j|j|||\}}}|jr!t|r9t|j } t | | d} | | ||d|_||_||_|j ||_dS)zInvoke the callback with a new snapshot Build the sntapshot from the current set of changes. Clear the current changes on completion. r1TN)_extract_changesrmrn_compute_snapshotrlrpr; functools cmp_to_keyrasortedr.rcclearrk) r(rZnext_resume_tokendeletesaddsupdates updated_tree updated_mapappliedChangesr1r.r)r)r*r's      z Watch.pushcCsg}g}g}|D]0\}}|tjkr||vr||q ||vr.|dur(||_||q |dur5||_||q |||fSr,)itemsrArCappendr)rmchangesrrrrrOr"r)r)r*rDs      zWatch._extract_changescs|}|}t|t|ksJdddddfdd}g} t|j} t|}|D]} | ||\} }}| | q-t|| d}td |D]} td | ||\} }}| | qKt|| d}|D]} || ||\} }}| dur}| | qit|t|ksJd ||| fS) NzJThe document tree and document map should have the same number of entries.cSsP||vsJd||}||}|j}||}||=ttj||d||fS)z Applies a document delete to the document tree and document map. Returns the corresponding DocumentChange event. z!Document to delete does not existr N)rr5r#r7rFrArC)rOrr old_documentexistingrIr)r)r* delete_docds   z+Watch._compute_snapshot..delete_doccSsN|jj}||vs Jd||d}||j}|||<ttj|d|||fS)z Applies a document add to the document tree and the document map. Returns the corresponding DocumentChange event. zDocument to add already existsNr )rrr2r5r#rFrArB) new_documentrrrOrJr)r)r*add_docvs  z(Watch._compute_snapshot..add_doccsv|jj}||vs Jd||}|j|jkr6|||\}}}|||\}}}ttj||j|j||fSd||fS)z Applies a document modification to the document tree and the document map. Returns the DocumentChange event for successful modifications. z!Document to modify does not existN) rrrrrFrArDrIrJ)rrrrOrZ remove_changeZ add_changerrr)r* modify_docs(     z+Watch._compute_snapshot..modify_docrzwalk over add_changeszin add_changeszQThe update document tree and document map should have the same number of entries.)r;rrrarrrr)r(rlrmZdelete_changesZ add_changesZupdate_changesrrrrr1rOZchangerNr)rr*rYsH!          zWatch._compute_snapshotcCs2||j|jd\}}}t|jt|t|S)zsReturn the current count of all documents. Count includes the changes from the current changeMap. N)rrmrnr;)r(rrrr)r)r*rszWatch._current_sizecCsHtd|jd|_|jD] }|jj}t j |j|<qd|_ dS)zG Helper to clear the docs on RESET or filter mismatch. zresetting documentsNF) rrrnrrkrlr.rrrArCro)r(rNrOr)r)r*rs   zWatch._reset_docsr,)#r>r?r@r+rq classmethodrrrvripropertyrrr{rrrrrrTargetChangeTypeZ NO_CHANGEZADDZREMOVERESETZCURRENTrrr|r staticmethodrrrrr)r)r)r*r^sB>     % q q r^)1 __future__r collectionsrloggingreenumrrRZgoogle.api_corerZgoogle.api_core.bidirrZgoogle.cloud.firestore_v1rZ)google.cloud.firestore_v1.types.firestorerr r r getLoggerr>rrZGRPC_STATUS_CODErZAbortedZ CancelledUnknownZDeadlineExceededZResourceExhaustedZInternalServerErrorZServiceUnavailableZUnauthenticatedrWr\ namedtupler!objectr$rArFrMrTrUrZr]r^r)r)r)r*s           +