o  ik@sdZddlZddlZddlZddlZddlZddlZddlm Z ddl m Z e e ZdZGdddeZGdd d eZGd d d e Zd d ZGdddeZGdddeZdS)z5Helpers for synchronous bidirectional streaming RPCs.N) exceptions) BidiRpcBasez!Thread-ConsumeBidirectionalStreamc@s*eZdZdZd ddZddZdd ZdS) _RequestQueueGeneratoraA helper for sending requests to a gRPC stream from a Queue. This generator takes requests off a given queue and yields them to gRPC. This helper is useful when you have an indeterminate, indefinite, or otherwise open-ended set of requests to send through a request-streaming (or bidirectional) RPC. Example:: requests = request_queue_generator(q) call = stub.StreamingRequest(iter(requests)) requests.call = call for response in call: print(response) q.put(...) Args: queue (queue_module.Queue): The request queue. period (float): The number of seconds to wait for items from the queue before checking if the RPC is cancelled. In practice, this determines the maximum amount of time the request consumption thread will live after the RPC is cancelled. initial_request (Union[protobuf.Message, Callable[None, protobuf.Message]]): The initial request to yield. This is done independently of the request queue to allow fo easily restarting streams that require some initial configuration request. NcCs||_||_||_d|_dSN)_queue_period_initial_requestcall)selfqueueperiodinitial_requestrt/var/www/snowflake_co_dev_github/snow_flake_back_end_deploy/env/lib/python3.10/site-packages/google/api_core/bidi.py__init__As z_RequestQueueGenerator.__init__cCs|jdup |jSrr is_activer rrr _is_activeGsz!_RequestQueueGenerator._is_activeccs|jdurt|jr|Vn|jV z |jj|jd}Wntjy6|s4t dYdSYqw|durBt ddS|sS|j |t ddS|Vq)NT)timeoutz9Empty queue and inactive call, exiting request generator.z"Cleanly exiting request generator.zEInactive call, replacing item on queue and exiting request generator.) r callablergetr queue_moduleEmptyr_LOGGERdebugput)r itemrrr__iter__Ns6     z_RequestQueueGenerator.__iter__)rN)__name__ __module__ __qualname____doc__rrrrrrrrs  ! rc@s0eZdZdZddZddZddZdd Zd S) _ThrottleaA context manager limiting the total entries in a sliding time window. If more than ``access_limit`` attempts are made to enter the context manager instance in the last ``time window`` interval, the exceeding requests block until enough time elapses. The context manager instances are thread-safe and can be shared between multiple threads. If multiple requests are blocked and waiting to enter, the exact order in which they are allowed to proceed is not determined. Example:: max_three_per_second = _Throttle( access_limit=3, time_window=datetime.timedelta(seconds=1) ) for i in range(5): with max_three_per_second as time_waited: print("{}: Waited {} seconds to enter".format(i, time_waited)) Args: access_limit (int): the maximum number of entries allowed in the time window time_window (datetime.timedelta): the width of the sliding time window cCsN|dkrtd|tdkrtd||_||_tj|d|_t |_ dS)Nrz&access_limit argument must be positiverz1time_window argument must be a positive timedelta)maxlen) ValueErrordatetime timedelta _time_window _access_limit collectionsdeque _past_entries threadingLock _entry_lock)r access_limit time_windowrrrrsz_Throttle.__init__cCs|j\tj|j}|jr%|jd|kr%|j|jr%|jd|kst|j|jkr?|jtj WddS|jd| }t ||jtj|WdS1sbwYdS)Nrg) r0r'nowr)r-popleftlenr*append total_secondstimesleep)r cutoff_timeto_waitrrr __enter__s  $z_Throttle.__enter__cGsdSrr)r _rrr__exit__sz_Throttle.__exit__cCsd|jj|jt|jS)Nz#{}(access_limit={}, time_window={}))format __class__r r*reprr)rrrr__repr__sz_Throttle.__repr__N)r r!r"r#rr<r>rBrrrrr$s  r$c@sDeZdZdZddZddZddZdd Zd d Ze d d Z dS)BidiRpca8A helper for consuming a bi-directional streaming RPC. This maps gRPC's built-in interface which uses a request iterator and a response iterator into a socket-like :func:`send` and :func:`recv`. This is a more useful pattern for long-running or asymmetric streams (streams where there is not a direct correlation between the requests and responses). Example:: initial_request = example_pb2.StreamingRpcRequest( setting='example') rpc = BidiRpc( stub.StreamingRpc, initial_request=initial_request, metadata=[('name', 'value')] ) rpc.open() while rpc.is_active(): print(rpc.recv()) rpc.send(example_pb2.StreamingRpcRequest( data='example')) rpc.close() This does *not* retry the stream on errors. See :class:`ResumableBidiRpc`. Args: start_rpc (grpc.StreamStreamMultiCallable): The gRPC method used to start the RPC. initial_request (Union[protobuf.Message, Callable[None, protobuf.Message]]): The initial request to yield. This is useful if an initial request is needed to start the stream. metadata (Sequence[Tuple(str, str)]): RPC metadata to include in the request. cCstS)zCreate a queue for requests.)rQueuerrrr _create_queueszBidiRpc._create_queuec Cs|jrtdt|j|jd}z |jt||jd}Wntj y/}z| |j d}~ww||_ t |dr@|j|j n||j ||_||_ dS)zOpens the stream.z#Cannot open an already open stream.)r)metadataN_wrapped)rr&r_request_queuer _start_rpciter _rpc_metadatarGoogleAPICallError _on_call_doneresponser hasattrrGadd_done_callback_request_generator)r request_generatorr excrrropens$    z BidiRpc.opencCs:|jdurdS|jd|jd|_d|_g|_dS)zCloses the stream.N)r rHrcancelrQr _callbacksrrrrcloses    z BidiRpc.closecCs:|jdur td|jr|j|dSt|jdS)zQueue a message to be sent on the stream. Send is non-blocking. If the underlying RPC has been closed, this will raise. Args: request (protobuf.Message): The request to send. Nz8Cannot send on an RPC stream that has never been opened.)r r&rrHrnextr requestrrrsend's  z BidiRpc.sendcCs|jdur tdt|jS)zWait for a message to be returned from the stream. Recv is blocking. If the underlying RPC has been closed, this will raise. Returns: protobuf.Message: The received message. Nz8Cannot recv on an RPC stream that has never been opened.)r r&rXrrrrrecv<s  z BidiRpc.recvcC|jduo |jS)z1True if this stream is currently open and active.NrrrrrrKzBidiRpc.is_activeN) r r!r"r#rErTrWr[r\propertyrrrrrrCs( rCcCsdS)z-By default, no errors cause BiDi termination.Fr)future_or_errorrrr_never_terminateQsracseZdZdZedddffdd ZddZdd Zd d Zd d Z ddZ ddZ ddZ ddZ fddZeddZZS)ResumableBidiRpca5A :class:`BidiRpc` that can automatically resume the stream on errors. It uses the ``should_recover`` arg to determine if it should re-establish the stream on error. Example:: def should_recover(exc): return ( isinstance(exc, grpc.RpcError) and exc.code() == grpc.StatusCode.UNAVAILABLE) initial_request = example_pb2.StreamingRpcRequest( setting='example') metadata = [('header_name', 'value')] rpc = ResumableBidiRpc( stub.StreamingRpc, should_recover=should_recover, initial_request=initial_request, metadata=metadata ) rpc.open() while rpc.is_active(): print(rpc.recv()) rpc.send(example_pb2.StreamingRpcRequest( data='example')) Args: start_rpc (grpc.StreamStreamMultiCallable): The gRPC method used to start the RPC. initial_request (Union[protobuf.Message, Callable[None, protobuf.Message]]): The initial request to yield. This is useful if an initial request is needed to start the stream. should_recover (Callable[[Exception], bool]): A function that returns True if the stream should be recovered. This will be called whenever an error is encountered on the stream. should_terminate (Callable[[Exception], bool]): A function that returns True if the stream should be terminated. This will be called whenever an error is encountered on the stream. metadata Sequence[Tuple(str, str)]: RPC metadata to include in the request. throttle_reopen (bool): If ``True``, throttling will be applied to stream reopen calls. Defaults to ``False``. NFcsbtt||||||_||_t|_d|_t |_ |r,t dt j ddd|_dSd|_dS)NF )seconds)r1r2)superrbr_should_recover_should_terminater.RLock_operational_lock _finalizedr/_finalize_lockr$r'r(_reopen_throttle)r start_rpcshould_recovershould_terminaterrFthrottle_reopenr@rrrs     zResumableBidiRpc.__init__cCs^|j"|jr WddS|jD]}||qd|_WddS1s(wYdS)NT)rlrkrV)r resultcallbackrrr _finalizes  "zResumableBidiRpc._finalizecCs|j8||r||n||s||ntd|WddSWddSWddS1s>wYdS)Nz%Re-opening stream from gRPC callback.)rjrhrurgrr_reopenr futurerrrrMs      "zResumableBidiRpc._on_call_donec Cs|jf|jdur|jrtd WddSd|_d|_z!|jr>|j |Wdn1s8wYn|Wnty[}z td|| |d}~wwt dWddS1slwYdS)Nz"Stream was already re-established.z"Failed to re-open stream due to %szRe-established stream) rjr rrrrQrmrT Exceptionruinfo)r rSrrrrvs.      "zResumableBidiRpc._reopenc Os z||i|WStyy}zd|jStd||||r?|td|||| WdWYd}~dS||sV|td|||||td||Wdn1sjwYWYd}~nd}~wwq)aWWraps a method to recover the stream and retry on error. If a retryable error occurs while making the call, then the stream will be re-opened and the method will be retried. This happens indefinitely so long as the error is a retryable one. If an error occurs while re-opening the stream, then this method will raise immediately and trigger finalization of this object. Args: method (Callable[..., Any]): The method to call. args: The args to pass to the method. kwargs: The kwargs to pass to the method. TzCall to retryable %r caused %s.zTerminating %r due to %s.NzNot retrying %r due to %s.z$Re-opening stream from retryable %r.) ryrjrrrhrWrurgrv)r methodargskwargsrSrrr _recoverables.      zResumableBidiRpc._recoverablecCs`|j |j}Wdn1swY|durtd|r*|j|dSt|dS)Nz1Cannot send on an RPC that has never been opened.)rjr r&rrHrrX)r rZr rrr_sends  zResumableBidiRpc._sendcCs||j|Sr)r~rrYrrrr[szResumableBidiRpc.sendcCsD|j |j}Wdn1swY|durtdt|S)Nz1Cannot recv on an RPC that has never been opened.)rjr r&rX)r r rrr_recvs zResumableBidiRpc._recvcCs ||jSr)r~rrrrrr\"s zResumableBidiRpc.recvcs|dtt|dSr)rurfrbrWrrrrrrW%s zResumableBidiRpc.closecCs<|j|jduo |j WdS1swYdS)z7bool: True if this stream is currently open and active.N)rjr rkrrrrr)s $zResumableBidiRpc.is_active)r r!r"r#rarrurMrvr~rr[rr\rWr_r __classcell__rrrrrrbVs$6  #% rbc@sbeZdZdZdddZddZddZd d Zd d Ze d dZ ddZ ddZ e ddZ dS)BackgroundConsumerauA bi-directional stream consumer that runs in a separate thread. This maps the consumption of a stream into a callback-based model. It also provides :func:`pause` and :func:`resume` to allow for flow-control. Example:: def should_recover(exc): return ( isinstance(exc, grpc.RpcError) and exc.code() == grpc.StatusCode.UNAVAILABLE) initial_request = example_pb2.StreamingRpcRequest( setting='example') rpc = ResumeableBidiRpc( stub.StreamingRpc, initial_request=initial_request, should_recover=should_recover) def on_response(response): print(response) consumer = BackgroundConsumer(rpc, on_response) consumer.start() Note that error handling *must* be done by using the provided ``bidi_rpc``'s ``add_done_callback``. This helper will automatically exit whenever the RPC itself exits and will not provide any error details. Args: bidi_rpc (BidiRpc): The RPC to consume. Should not have been ``open()``ed yet. on_response (Callable[[protobuf.Message], None]): The callback to be called for every response on the stream. on_fatal_exception (Callable[[Exception], None]): The callback to be called on fatal errors during consumption. Default None. NcCs6||_||_d|_||_t|_d|_t|_ dS)NF) _bidi_rpc _on_response_paused_on_fatal_exceptionr. Condition_wake_threadr/rj)r bidi_rpc on_responseon_fatal_exceptionrrrr_s zBackgroundConsumer.__init__cCs |dSr)resumerwrrrrMhs z BackgroundConsumer._on_call_donec CsXz[||j|j|j|jjrZ|j|jr.t d|j t d|jsWdn1s8wYt d|j }t d|j durV| ||jjsWnHt jy}ztj dt|dd|jdurw||WYd}~n'd}~wty}ztdt||jdur||WYd}~nd}~wwtd tdS) Nzpaused, waiting for waking.zwoken.zwaiting for recv.zrecved response.z%s caught error %s and will exit. Generally this is due to the RPC itself being cancelled and the error will be surfaced to the calling code.T)exc_infoz0%s caught unexpected exception %s and will exit.z %s exiting)setrrPrMrTrrrrrwaitr\rrrL_BIDIRECTIONAL_CONSUMER_NAMErry exceptionrz)r readyrNrSrrr _thread_mainmsR               zBackgroundConsumer._thread_maincCsn|j*t}tjt|j|fdd}||||_t d|j WddS1s0wYdS)z;Start the background thread and begin consuming the thread.T)nametargetr|daemonzStarted helper thread %sN) rjr.EventThreadrrstartrrrrr)r rthreadrrrrs"zBackgroundConsumer.startcCsz|j0|j|jdur"||jd|jr"tdd|_d|_ d|_ WddS1s6wYdS)zStop consuming the stream and shutdown the background thread. NOTE: Cannot be called within `_thread_main`, since it is not possible to join a thread to itself. Ng?zBackground thread did not exit.) rjrrWrrjoinis_aliverwarningrrrrrrstops     "zBackgroundConsumer.stopcCr])z.bool: True if the background thread is active.N)rrrrrrrr^zBackgroundConsumer.is_activecCs2|j d|_WddS1swYdS)zWPauses the response stream. This does *not* pause the request stream. TN)rrrrrrpauses"zBackgroundConsumer.pausecCs<|jd|_|jWddS1swYdS)zResumes the response stream.FN)rr notify_allrrrrrs "zBackgroundConsumer.resumecCs|jS)z,bool: True if the response stream is paused.)rrrrr is_pausedszBackgroundConsumer.is_pausedr)r r!r"r#rrMrrrr_rrrrrrrrr7s ' 3 r)r#r+r'loggingr rr.r8google.api_corergoogle.api_core.bidi_baser getLoggerr rrobjectrr$rCrarbrrrrrs&   oCb