a !fec@szddlZddlmZddlZddlZddlZddlmZddlmZddl m Z ddl Z ddl m Z ddl mZddl mZdd lmZejZeeZd Zdd d d dddddddddddddddZdZe je je je je je je je j fZ!e jfZ"e#ddd gZ$Gd!d"d"e%Z&Gd#d$d$eZ'Gd%d&d&e%Z(Gd'd(d(e%Z)d)d*Z*d+d,Z+d-d.Z,d/d0Z-Gd1d2d2e%Z.dS)3N)Enum)ResumableBidiRpc)BackgroundConsumer) exceptions) ListenRequest)Target) TargetChange)_helpersiyP )OK CANCELLEDUNKNOWNZINVALID_ARGUMENTZDEADLINE_EXCEEDED NOT_FOUNDZALREADY_EXISTSZPERMISSION_DENIEDZUNAUTHENTICATEDZRESOURCE_EXHAUSTEDZFAILED_PRECONDITIONZABORTEDZ OUT_OF_RANGEZ UNIMPLEMENTEDZINTERNAL UNAVAILABLEZ DATA_LOSSZ DO_NOT_USEzThread-OnRpcTerminated DocTreeEntryvalueindexc@sTeZdZddZddZddZddZd d Zd d Zd dZ ddZ ddZ dS) WatchDocTreecCsi|_d|_dS)Nr)_dict_indexselfr(`/var/www/html/python-backend/venv/lib/python3.9/site-packages/google/cloud/firestore_v1/watch.py__init__NszWatchDocTree.__init__cCst|jSN)listr$keysr&r(r(r)r-RszWatchDocTree.keyscCs"t}|j|_|j|_|}|Sr+)r#r$copyr%)r'Zwdtr(r(r)_copyUs  zWatchDocTree._copycCs,|}t||j|j|<|jd7_|S)Nr )r/r r%r$)r'keyr!r(r(r)insert\szWatchDocTree.insertcCs |j|Sr+r$r'r0r(r(r)findbszWatchDocTree.findcCs|}|j|=|Sr+)r/r$r3r(r(r)removeeszWatchDocTree.removeccs|jD] }|VqdSr+r2r'kr(r(r)__iter__js zWatchDocTree.__iter__cCs t|jSr+)lenr$r&r(r(r)__len__nszWatchDocTree.__len__cCs ||jvSr+r2r6r(r(r) __contains__qszWatchDocTree.__contains__N) __name__ __module__ __qualname__r*r-r/r1r4r5r8r:r;r(r(r(r)r#Jsr#c@seZdZdZdZdZdS) ChangeTyper r r N)r<r=r>ADDEDREMOVEDMODIFIEDr(r(r(r)r?usr?c@seZdZddZdS)DocumentChangecCs||_||_||_||_dS)zDocumentChange Args: type (ChangeType): document (document.DocumentSnapshot): old_index (int): new_index (int): N)typedocument old_index new_index)r'rDrErFrGr(r(r)r*|s zDocumentChange.__init__Nr<r=r>r*r(r(r(r)rC{srCc@seZdZddZdS) WatchResultcCs||_||_||_dSr+)snapshotname change_type)r'rJrKrLr(r(r)r*szWatchResult.__init__NrHr(r(r(r)rIsrIcCst|tjrt|S|S)z(Wraps a gRPC exception class, if needed.) isinstancegrpcZRpcErrorrZfrom_grpc_error) exceptionr(r(r)_maybe_wrap_exceptions  rPcCs||ksJddS)Nz+Document watches only support one document.rr()Zdoc1Zdoc2r(r(r)document_watch_comparatorsrQcCst|}t|tSr+)rPrM_RECOVERABLE_STREAM_EXCEPTIONSrOwrappedr(r(r)_should_recoversrUcCst|}t|tSr+)rPrM_TERMINATING_STREAM_EXCEPTIONSrSr(r(r)_should_terminatesrWc @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_requestrr^ _transportlistenrUrWrYZ _rpc_metadata_rpcadd_done_callback _on_rpc_doner on_snapshot _consumerstart)r'Z rpc_requestr(r(r)rkszWatch._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)ry target_id)_client_document_pathWATCH_TARGET_IDrQ)cls document_refrmrnr(r(r) for_documents zWatch.for_documentcCs>|j\}}tj||d}|||j|jtd|j||S)N)parentZstructured_query)queryrz) _parentZ _parent_inforZ QueryTargetZ _to_protobufr{_pbr}r[)r~rrmrn parent_path_Z query_targetr(r(r) for_querys zWatch.for_querycCs8|jdur|j|jd<n|jddt|jj|jdS)Nre)ZdatabaseZ add_target)rerZpoprrYrdr&r(r(r)rp)s   zWatch._get_rpc_requestcCs|d|_t|j|_dS)Nz /documents/)_documents_pfxr9_documents_pfx_len)r'Zdatabase_stringr(r(r)rc3s 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)rw is_activer&r(r(r)r7szWatch.is_activeNcCs|jf|jrWddS|jr8td|jd|_|jd|_d|_tdWdn1sr0Y|rtd|t |t r|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) rarbr_LOGGERdebugrwstoprscloserM Exception RuntimeError)r'reasonr(r(r)r@s    ( 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)rKrlkwargsTN) rinforPr_Thread_RPC_ERROR_THREAD_NAMErdaemonrx)r'futurethreadr(r(r)ru_s  zWatch._on_rpc_donecCs |dSr+)rr&r(r(r) unsubscriberszWatch.unsubscribecCsFtd|jdup t|jdk}|rB|jrB|jrB||j|jdS)Nz%on_snapshot: target change: NO_CHANGEr)rr target_idsr9 read_timeripushre)r' target_changeZ no_target_idsr(r(r)$_on_snapshot_target_change_no_changeus  z*Watch._on_snapshot_target_change_no_changecCs,td|jd}|tkr(td|dS)Nzon_snapshot: target change: ADDrz&Unexpected target ID %s sent by server)rrrr}r)r'rrzr(r(r)_on_snapshot_target_change_adds  z$Watch._on_snapshot_target_change_addcCsDtd|jjr$|jj}|jj}nd}d}d||f}t|dS)Nz"on_snapshot: target change: REMOVErzinternal errorz Error %s: %s)rrcausecodemessager)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)rrrirr(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 Cs`|dur|dS|j}|d}|dkr|jj}td||j|}|durd|}t d||jt |dz|||jWn6t y}ztd|WYd}~n d}~00n|d krtd t |j jv}t |j jv} |j j} |rbtd t| j|j} || j} |j| } |j| | d d| j| jd }||j| j<n| r\tdtj|j| j<n|dkrtd|jj}tj|j|<n|dkrtd|jj}tj|j|<n|dkr8td|jj |!kr\t dt"j#t$|jd}|%|&|'|(n$tdd|}|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.)rKrlzUNKNOWN TYPE. UHOHzUnknown listen response type: ))rrZ WhichOneofrtarget_change_typerr_target_changetype_dispatchgetr ValueErrorrr}rrZremoved_target_idsrEr Z decode_dictfieldsrYrrKr\rrrhr?rArrrcount _current_sizer_rrrxjoinrrk)r'protoZpbwhichrmethrZexc2changedremovedrErrrrJrKrr(r(r)rvs~                      zWatch.on_snapshotc Cs||j|j|\}}}||j|j|||\}}}|jrBt|rrt|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. r0TN)_extract_changesrgrh_compute_snapshotrfrjr9 functools cmp_to_keyr[sortedr-r]clearre) r'rZnext_resume_tokendeletesaddsupdates updated_tree updated_mapappliedChangesr0r-r(r(r)rs    z Watch.pushcCsg}g}g}|D]`\}}|tjkr:||vrt||q||vr\|durP||_||q|durj||_||q|||fSr+)itemsr?rAappendr)rgchangesrrrrrKr!r(r(r)r8s    zWatch._extract_changescs |}|}t|t|ks Jdddddfdd}g} t|j} t|}|D] } | ||\} }}| | qZt|| d}td |D]*} td | ||\} }}| | qt|| d}|D](} || ||\} }}| dur| | qt|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)rr4r"r5rCr?rA)rKrr old_documentexistingrFr(r(r) delete_docXs   z+Watch._compute_snapshot..delete_doccSsN|jj}||vsJd||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)rr|r1r4r"rCr?r@) new_documentrrrKrGr(r(r)add_docjs  z(Watch._compute_snapshot..add_doccsv|jj}||vsJd||}|j|jkrl|||\}}}|||\}}}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) rr|rrrCr?rBrFrG)rrrrKrZ remove_changeZ add_changerrr(r) modify_doczs(     z+Watch._compute_snapshot..modify_docrzwalk over add_changeszin add_changeszQThe update document tree and document map should have the same number of entries.)r9rrr[rrrr)r'rfrgZdelete_changesZ add_changesZupdate_changesrrrrr0rKZchangerJr(rr)rMsF!           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)rrgrhr9)r'rrrr(r(r)rszWatch._current_sizecCsHtd|jd|_|jD]}|jj}t j |j|<q$d|_ dS)zG Helper to clear the docs on RESET or filter mismatch. zresetting documentsNF) rrrhrrerfr-rr|r?rAri)r'rJrKr(r(r)rs  zWatch._reset_docs)N)#r<r=r>r*rk classmethodrrrprcpropertyrrrurrrrrrTargetChangeType NO_CHANGEZADDZREMOVERESETZCURRENTrrrvr staticmethodrrrrr(r(r(r)rXs@>      o qrX)/ collectionsenumrrloggingr_Zgoogle.api_core.bidirrZgoogle.api_corerrNZ)google.cloud.firestore_v1.types.firestorerrrZgoogle.cloud.firestore_v1r r getLoggerr<rr}ZGRPC_STATUS_CODErZAbortedZ CancelledUnknownZDeadlineExceededZResourceExhaustedZInternalServerErrorZServiceUnavailableZUnauthenticatedrRrV namedtupler objectr#r?rCrIrPrQrUrWrXr(r(r(r)sn          +