gmdZddlZddlZddlZddlZddlZddlZddlm Z eje Z dZ GddeZGddeZGd d eZd ZGd d eZGddeZy)z%Bi-directional streaming RPC helpers.N) exceptionsz!Thread-ConsumeBidirectionalStreamc$eZdZdZddZdZdZy)_RequestQueueGeneratora1 A 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. The reason this is necessary is because gRPC takes an iterator as the request for request-streaming RPCs. gRPC consumes this iterator in another thread to allow it to block while generating requests for the stream. However, if the generator blocks indefinitely gRPC will not be able to clean up the thread as it'll be blocked on `next(iterator)` and not be able to check the channel status to stop iterating. This helper mitigates that by waiting on the queue with a timeout and checking the RPC state before yielding. Finally, it allows for retrying without swapping queues because if it does pull an item off the queue when the RPC is inactive, it'll immediately put it back and then exit. This is necessary because yielding the item in this case will cause gRPC to discard it. In practice, this means that the order of messages is not guaranteed. If such a thing is necessary it would be easy to use a priority queue. Example:: requests = request_queue_generator(q) call = stub.StreamingRequest(iter(requests)) requests.call = call for response in call: print(response) q.put(...) Note that it is possible to accomplish this behavior without "spinning" (using a queue timeout). One possible way would be to use more threads to multiplex the grpc end event with the queue, another possible way is to use selectors and a custom event/queue object. Both of these approaches are significant from an engineering perspective for small benefit - the CPU consumed by spinning is pretty minuscule. 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. Nc<||_||_||_d|_yN)_queue_period_initial_requestcall)selfqueueperiodinitial_requests q/var/www/html/FastMealFinder_FlaskServer-InitialRelease/venv/lib/python3.12/site-packages/google/api_core/bidi.py__init__z_RequestQueueGenerator.__init__Us  / cV|jduxs|jjSrr is_activer s r _is_activez!_RequestQueueGenerator._is_active[s% yyD 9DII$7$7$99rc#K|j6t|jr|jn|j |jj|j}|tjdy|js1|jj|tjdy|#t j $r*|jstjdYyYwxYww)N)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 items r__iter__z_RequestQueueGenerator.__iter__bs  ,--.++--+++ {{t||<| BC??$ % )J; %% (MMV s+AD &C -AD  8DD DD )N)__name__ __module__ __qualname____doc__rrr"rrrrs4l :$rrc(eZdZdZdZdZdZdZy) _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 c|dkr td|tjdkr td||_||_t j ||_tj|_ y)Nr#z&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_windows rrz_Throttle.__init__sl ! EF F (,,Q/ /PQ Q')(.. %>>+rc|j5tjj|jz }|jrK|jd|kr9|jj |jr|jd|kr9t |j|jkrA|jjtjj dddy|jd|z j}tj||jjtjj|cdddS#1swYyxYw)Nrg) r7r.nowr0r4popleftlenr1append total_secondstimesleep)r cutoff_timeto_waits r __enter__z_Throttle.__enter__s(    "++//1D4E4EEK$$););A)>)L""**,$$););A)>)L4%%&););;""))(*;*;*?*?*AB  ))!,{:IIKG JJw     % %h&7&7&;&;&= >   sBE*AE*3A-E**E3cyrr()r _s r__exit__z_Throttle.__exit__s rcdj|jj|jt |j S)Nz#{}(access_limit={}, time_window={}))format __class__r$r1reprr0rs r__repr__z_Throttle.__repr__s74;; NN # #T%7%7d>O>O9P  rN)r$r%r&r'rrDrGrLr(rrr*r*s2 ,$  rr*c\eZdZdZd dZdZdZdZdZdZ d Z e d Z e d Z y) BidiRpca#A 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')) 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. Nc||_||_||_tj|_d|_d|_g|_d|_ yNF) _start_rpcr _rpc_metadatarQueue_request_queue_request_generatorr _callbacksr )r start_rpcrmetadatas rrzBidiRpc.__init__sG# /%*002"& rc:|jj|y)a{Adds a callback that will be called when the RPC terminates. This occurs when the RPC errors or is successfully terminated. Args: callback (Callable[[grpc.Future], None]): The callback to execute. It will be provided with the same gRPC future as the underlying stream which will also be a :class:`grpc.Call`. N)rVr>)r callbacks radd_done_callbackzBidiRpc.add_done_callbacks x(rc6|jD] }|| yr)rV)r futurerZs r _on_call_donezBidiRpc._on_call_done s  H V  rc|jr tdt|j|j} |j t ||j}||_ t|dr&|jj|jn|j|j||_||_ y#tj$r!}|j|jd}~wwxYw)zOpens the stream.z$Can not open an already open stream.)r)rXN_wrapped)rr-rrTr rQiterrRrGoogleAPICallErrorr^responser hasattrr`r[rU)r request_generatorr excs ropenz BidiRpc.opens >>CD D2   1F1F  ??4(9#:TEWEW?XD"& 4 $ MM + +D,>,> ?  " "4#5#5 6"3 !,,    s|| ,   s&CC7C22C7c|jy|jjd|jjd|_d|_g|_y)zCloses the stream.N)r rTr cancelrUr rVrs rclosez BidiRpc.close-sJ 99   % "& $rc|j td|jjr|jj |yt |jy)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. N6Can not send() on an RPC that has never been open()ed.)r r-rrTr nextr requests rsendz BidiRpc.send:sM 99 UV V 99      # #G , OrcZ|j 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. 6Can not recv() on an RPC that has never been open()ed.)r r-rmrs rrecvz BidiRpc.recvOs( 99 UV VDIIrcV|jduxr|jjSz7bool: True if this stream is currently open and active.Nrrs rrzBidiRpc.is_active^s%yy$>)<)<)>>rc6|jjS)z:int: Returns an estimate of the number of queued requests.)rTqsizers rpending_requestszBidiRpc.pending_requestscs""((**r)NN)r$r%r&r'rr[r^rgrjrprspropertyrrxr(rrrNrNsS$L )8 * ??++rrNcy)z-By default, no errors cause BiDi termination.Fr()future_or_errors r_never_terminater|is rcveZdZdZedddffd ZdZdZdZdZ d Z d Z d Z d Z fd ZedZxZS)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``. NFctt| |||||_||_t j |_d|_t j|_ |r'tdtjd|_yd|_y)NF )seconds)r8r9)superr~r_should_recover_should_terminater5RLock_operational_lock _finalizedr6_finalize_lockr*r.r/_reopen_throttle)r rWshould_recovershould_terminaterrXthrottle_reopenrJs rrzResumableBidiRpc.__init__sx .y/8T-!1!*!2'nn. $-H,>,>r,J%D !%)D !rc|j5|jr dddy|jD] }|| d|_dddy#1swYyxYw)NT)rrrV)r resultrZs r _finalizezResumableBidiRpc._finalizesZ   # # #!OO !  !#DO # # #sA  A  Ac|j5|j|r|j|nH|j|s|j|n%tj d|j dddy#1swYyxYw)Nz%Re-opening stream from gRPC callback.)rrrrrr_reopenr r]s rr^zResumableBidiRpc._on_call_donesl # # %%f-v&))&1v& EF    s A,BB c&|j5|j9|jjrtj d dddyd|_d|_ |j r&|j 5|jdddn|jtjddddy#1swY'xYw#t$r-}tj d||j|d}~wwxYw#1swYyxYw)Nz"Stream was already re-established.z"Failed to re-open stream due to %szRe-established stream) rr rrrrUrrg Exceptionrinfo)r rfs rrzResumableBidiRpc._reopens  # # 2yy$)<)<)> BC 2 2 DI'+D # ((..$ $$IIK LL0 1A 2 2*$$   BCHs# 7 2 2sM=DD"C:C C$DC C D(C??DDDcJ ||i|S#t$r }|j5tjd|||j |rF|j tjd|||j | dddYd}~y|j|s:|j tjd|||j ||tjd||jdddn #1swYnxYwYd}~nd}~wwxYw#)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. zCall to retryable %r caused %s.zTerminating %r due to %s.NzNot retrying %r due to %s.z$Re-opening stream from retryable %r.) rrrrrrjrrr)r methodargskwargsrfs r _recoverablezResumableBidiRpc._recoverables #t.v.. #++#MM"CVSQ--c2  &A63Os+## //4  &BFCPs+! MM"H&QLLN!### # s9 D  DA"D DA1D  D D DD c|j5|j}ddd td|jr|jj |yt |y#1swYNxYw)Nrl)rr r-rrTr rm)r ror s r_sendzResumableBidiRpc._sendsj # # 99D  <UV V >>     # #G ,  J  s A''A0c:|j|j|Sr)rrrns rrpzResumableBidiRpc.send.s  W55rc|j5|j}ddd tdt|S#1swY!xYw)Nrr)rr r-rm)r r s r_recvzResumableBidiRpc._recv1sF  # # 99D  <UV VDz   s :Ac8|j|jSr)rrrs rrszResumableBidiRpc.recv:s  ,,rcL|jdtt|yr)rrr~rj)r rJs rrjzResumableBidiRpc.close=s t +-rc|j5|jduxr |j cdddS#1swYyxYwru)rr rrs rrzResumableBidiRpc.is_activeAs> # # A99D(@-@ A A As4=)r$r%r&r'r|rrr^rrrrprrsrjryr __classcell__)rJs@rr~r~nsa0l*).# !2F##J.6-. A Arr~cZeZdZdZdZdZdZdZdZe dZ dZ d Z e d Z y ) BackgroundConsumeraA 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. c||_||_d|_tj|_d|_tj|_yrP) _bidi_rpc _on_response_pausedr5 Condition_wake_threadr6r)r bidi_rpc on_responses rrzBackgroundConsumer.__init__us>!' ((*  !*!1rc$|jyr)resumers rr^z BackgroundConsumer._on_call_done}s  rc |j|jj|j|jj |jj r|j 5|jrQtjd|j jtjd|jrQdddtjd|jj}tjd|j||jj rtj%d ty#1swYxYw#tj$r'}tjdt|dYd}~\d}~wt $r%}tj#dt|Yd}~d}~wwxYw) 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)setrr[r^rgrrrrrwaitrsrrrb_BIDIRECTIONAL_CONSUMER_NAMEr exceptionr)r readyrcrfs r _thread_mainzBackgroundConsumer._thread_mains[)  IIK NN , ,T-?-? @ NN   !..**ZZ0,, &CD ) h/,,0  12>>..0 01!!(+'..**L  \#?@900,,  MM0-       B,   s>A1E,3AE A3E, E)%E,,G?F!! G-G  GcX|j5tj}tjt|j |f}d|_|j|j||_ tjd|jdddy#1swYyxYw)z;Start the background thread and begin consuming the thread.)nametargetrTzStarted helper thread %sN) rr5EventThreadrrdaemonstartrrrrr)r rthreads rrzBackgroundConsumer.starts  # # COO%E%%1((XF !FM LLN JJL!DL MM4fkk B C C Cs B B  B)cb|j5|jj|jZ|j |jj d|jj rtjdd|_d|_ dddy#1swYyxYw)z=Stop consuming the stream and shutdown the background thread.Ng?zBackground thread did not exit.) rrrjrrjoinis_aliverwarningrrs rstopzBackgroundConsumer.stops  # # % NN "||'  !!#&<<((*OO$EFDL $D  % % %s BB%%B.cV|jduxr|jjS)z.bool: True if the background thread is active.N)rrrs rrzBackgroundConsumer.is_actives%||4'CDLL,A,A,CCrcT|j5d|_dddy#1swYyxYw)zWPauses the response stream. This does *not* pause the request stream. TN)rrrs rpausezBackgroundConsumer.pauses' ZZ DL   s'c|j5d|_|jjdddy#1swYyxYw)zResumes the response stream.FN)rr notify_allrs rrzBackgroundConsumer.resumes5 ZZ $ DL JJ ! ! # $ $ $s "8Ac|jS)z,bool: True if the response stream is paused.)rrs r is_pausedzBackgroundConsumer.is_pauseds||rN)r$r%r&r'rr^rrrryrrrrr(rrrrOsX#J2 ,A\C&%"DD $ rr)r'r2r.loggingr rr5r@google.api_corer getLoggerr$rrobjectrr*rNr|r~rr(rrrs, & '  H %BhVhV@ @ FZ+fZ+z ^Aw^AB[[r