o $iw @sddlZddlZddlZddlZddlZddlZddlmZddlm Z ddl m Z ddl m Z mZmZmZmZmZmZmZmZddlZddlZddlmZddlmZddlmZdd lmZdd l m!Z!dd l"m#Z#m$Z$m%Z%m&Z&m'Z'dd l(m)Z)dd l*m+Z+ddl,m-Z-ddl.m/Z/ddl0m1Z1m2Z2ddl3m4Z4ddl5m6Z6m7Z7m8Z8m9Z9m:Z:ddl;mZ>m?Z?m@Z@ddlAmBZBddlCmDZDmEZEmFZFeGe:ZHe ddGdddZIdeJde%fddZK dPd ee d!eLdee$fd"d#ZMGd$d%d%ZNd&d'ZOGd(d)d)e'ZPGd*d+d+ZQd,eddfd-d.ZRd/ejdeSfd0d1ZTGd2d3d3ZUd4e de!gfd5d6ZVGd7d8d8ZWd9e#d4eede#fd:d;ZXdeBd?ejYd@eSdejZf dAdBZ[dCe\dDe]dEe^deeBdeBfdLdMZad>eBdeBfdNdOZbdS)QN)deque)deepcopy) dataclass) AnyAsyncGenerator AwaitableCallableListOptionalTupleTypeUnion)FastAPI)jsonable_encoder)version)MutableHeaders) Middleware)ASGIAppMessageReceiveScopeSend)Config) LifespanOn)is_ipv6) IS_PYDANTIC_2) RayActorError RayTaskError)RequestMetadata)#RAY_SERVE_HTTP_KEEP_ALIVE_TIMEOUT_S)RAY_SERVE_HTTP_PROXY_CALLBACK_IMPORT_PATH&RAY_SERVE_REQUEST_PROCESSING_TIMEOUT_SSERVE_HTTP_REQUEST_ID_HEADERSERVE_LOGGER_NAME)ResponseStatus)call_function_from_import_pathgenerate_request_idserve_encoders) HTTPOptions)BackPressureErrorDeploymentUnavailableErrorRayServeExceptionT)frozenc@sPeZdZUeed<eed<eed<deeeeffddZde j j fddZ d S) ASGIArgsscopereceivesendreturncCs|j|j|jfSN)r.r/r0selfr5Y/home/ubuntu/veenaModal/venv/lib/python3.10/site-packages/ray/serve/_private/http_util.py to_args_tupleBszASGIArgs.to_args_tuplecCstjj|Sr2) starletterequestsRequestr7r3r5r5r6to_starlette_requestEszASGIArgs.to_starlette_requestN) __name__ __module__ __qualname__r__annotations__rrr r7r8r9r:r;r5r5r5r6r-<s r-serialized_bodyr1csdfdd}|S)zAReturns an ASGI receiver that returns the provided buffered body.Fcs,rt}|IdHddddS)NT http.requestF)bodytype more_body)asyncioEventwait) block_foreverreceivedr@r5r6 mock_receiveRs  z0make_buffered_asgi_receive..mock_receiver5)r@rKr5rIr6make_buffered_asgi_receiveKs rLobj status_codecCsd}d}|dur d}d}n&t|tr|}d}nt|tr$|d}d}ntjt|tddd}d }d |d |ggd d |dgS)zSerializes the provided object and converts it to ASGI messages. These ASGI messages can be sent via an ASGI `send` interface to comprise an HTTP response. Ns text/plainzutf-8stext/plain; charset=utf-8)custom_encoder),:) separatorssapplication/jsonhttp.response.starts content-type)rCstatusheaderszhttp.response.body)rCrB) isinstancebytesstrencodejsondumpsrr')rNrOrB content_typer5r5r6convert_object_to_asgi_messagesbs0    r_c@s"eZdZdZdddZddZdS) ResponseaQASGI compliant response class. It is expected to be called in async context and pass along `scope, receive, send` as in ASGI spec. >>> from ray.serve.http_util import Response # doctest: +SKIP >>> scope, receive = ... # doctest: +SKIP >>> await Response({"k": "v"}).send(scope, receive, send) # doctest: +SKIP NrMcCst||d|_dS)zConstruct a HTTP Response based on input type. Args: content: Any JSON serializable object. status_code (int, optional): Default status code is 200. )rNrON)r_ _messages)r4contentrOr5r5r6__init__s zResponse.__init__cs |jD] }||IdHqdSr2)ra)r4r.r/r0messager5r5r6r0s z Response.sendNrM)r<r=r>__doc__rcr0r5r5r5r6r`s r`csNg}d}|r"|IdH}|ddksJ|d}||d|sd|S)NTrCrArDrBrP)appendjoin)r.r/r0 body_bufferrDrdr5r5r6receive_http_bodys  rjc@seZdZdZddZddZdefddZd efd d Z d efd d Z ddZ de efddZ defddZdejdee edffddZdS) MessageQueueaqQueue enables polling for received or sent messages. Implements the ASGI `Send` interface. This class: - Is *NOT* thread safe and should only be accessed from a single asyncio event loop. - Assumes a single consumer of the queue (concurrent calls to `get_messages_nowait` and `wait_for_message` is undefined behavior). cCs"t|_t|_d|_d|_dS)NF)r_message_queuerErF_new_message_event_closed_errorr3r5r5r6rcs  zMessageQueue.__init__cCsd|_|jdS)zClose the queue, rejecting new messages. Once the queue is closed, existing messages will be returned from `get_messages_nowait` and subsequent calls to `wait_for_message` will always return immediately. TN)rnrmsetr3r5r5r6closeszMessageQueue.closeecC ||_dSr2)ro)r4rrr5r5r6 set_error zMessageQueue.set_errorrdcCs|j||jdSr2)rlrgrmrpr4rdr5r5r6 put_nowaits zMessageQueue.put_nowaitcs|jrtd||dS)zSend a message, putting it on the queue. `RuntimeError` is raised if the queue has been closed using `.close()`. z6New messages cannot be sent after the queue is closed.N)rn RuntimeErrorrwrvr5r5r6__call__szMessageQueue.__call__cs |js|jIdHdSdS)a Wait until at least one new message is available. If a message is available, this method will return immediately on each call until `get_messages_nowait` is called. After the queue is closed using `.close()`, this will always return immediately. N)rnrmrGr3r5r5r6wait_for_messages zMessageQueue.wait_for_messager1cCs>g}t|jdkr||jt|jdks |j|S)aReturns all messages that are currently available (non-blocking). At least one message will be present if `wait_for_message` had previously returned and a subsequent call to `wait_for_message` blocks until at least one new message is available. r)lenrlrgpopleftrmclear)r4messagesr5r5r6get_messages_nowaits  z MessageQueue.get_messages_nowaitcs|jr|j|jIdHt|jdkr,|j}t|jdkr*|js*|j|St|jdkr9|jr9|jt|jdkrE|jrGtdSdS)aThis blocks until a message is ready. This method should not be used together with get_messages_nowait. Please use either `get_one_message` or `get_messages_nowait`. Raises: StopAsyncIteration: if the queue is closed and there are no more messages. Exception (self._error): if there are no more messages in the queue and an error has been set. Nr) rormrGr{rlr|rnr}StopAsyncIteration)r4msgr5r5r6get_one_messages   zMessageQueue.get_one_messagecall_futNcCsd}zL t|}tj||gtjdIdH\}}|}|r$|V||vr)nq|}|dur5|dW|s>||durL|sN|dSdSdS|sX||dure|sf|www)aRepeatedly consume messages from the queue and yield them. This is used to fetch queue messages in the system event loop in a thread-safe manner. Args: call_fut: The async Future pointing to the task from the user code event loop that is pushing messages onto the queue. Yields: List[Any]: Messages from the queue. NT) return_when) rE create_taskrzrGFIRST_COMPLETEDr exceptiondonecancel)r4rwait_for_msg_taskr_r~rrr5r5r6fetch_messages_from_queues8   z&MessageQueue.fetch_messages_from_queue)r<r=r>rfrcrq BaseExceptionrtrrwryrzr rrrEFuturerrrr5r5r5r6rks    rkc @sleZdZdZdededeegeeffddZ de fdd Z e de jfd d Zd d Zde fddZdS)ASGIReceiveProxyzProxies ASGI receive from an actor. The `receive_asgi_messages` callback will be called repeatedly to fetch messages until a disconnect message is received. r.request_metadatareceive_asgi_messagescCs&|d|_d|_||_||_d|_dS)NrC)_type_queue_request_metadata_receive_asgi_messages_disconnect_message)r4r.rrr5r5r6rcJs  zASGIReceiveProxy.__init__r1cCs|jdkr dddSddiS)a7Return the appropriate disconnect message based on the connection type. HTTP ASGI spec: https://asgi.readthedocs.io/en/latest/specs/www.html#disconnect-receive-event WS ASGI spec: https://asgi.readthedocs.io/en/latest/specs/www.html#disconnect-receive-event-ws websocketwebsocket.disconnecti)rCcoderChttp.disconnect)rr3r5r5r6_get_default_disconnect_messageWs z0ASGIReceiveProxy._get_default_disconnect_messagecCs|jdur t|_|jSr2)rrEQueuer3r5r5r6queueis  zASGIReceiveProxy.queuec s z&||jIdH}t|D]}|j||ddvr&||_WdSqWn/ty?|}|j|||_YdSt yW}z |j|WYd}~dSd}~wwq)aFetch messages repeatedly until a disconnect message is received. If a disconnect message is received, this function exits and returns it. If an exception occurs, it will be raised on the next __call__ and no more messages will be received. TNrC>rr) rrpickleloadsrrwrKeyErrorr Exception)r4pickled_messagesrdrrr5r5r6fetch_until_disconnectps0      z'ASGIReceiveProxy.fetch_until_disconnectcs>|jr|jdur|jS|jIdH}t|tr||S)zReturn the next message once available. This will repeatedly return a disconnect message once it's been received. N)remptyrgetrXrrvr5r5r6rys zASGIReceiveProxy.__call__N)r<r=r>rfrrrrrYrcrrpropertyrErrrryr5r5r5r6rCs   rclscsddlm}m}ddlmmdd}fdd|jD}|}|D]O}|j||j}t |} t | j } t| dkrFtd| d} | j||d } | gd d| d d D} | j| d }||j_|j_|j|q%||t |jD].}t|fsqtst|r|jr|jj|j_t|jdd }|d ur|kr|qfdd|jD|jd d <d S)a_Transform the `cls`'s methods and class annotations to FastAPI routes. Modified from https://github.com/dmontagu/fastapi-utils/blob/master/fastapi_utils/cbv.py Usage: >>> from fastapi import FastAPI >>> app = FastAPI() # doctest: +SKIP >>> class A: # doctest: +SKIP ... @app.route("/{i}") # doctest: +SKIP ... def func(self, i: int) -> str: # doctest: +SKIP ... return self.dep + i # doctest: +SKIP >>> # just running the app won't work, here. >>> make_fastapi_class_based_view(app, A) # doctest: +SKIP >>> # now app can be run properly r) APIRouterDepends)APIRouteAPIWebSocketRoutecsddlm}|jS)Nrserve)rayrget_replica_contextservable_objectrr5r5r6get_current_servable_instances  zDmake_fastapi_class_based_view..get_current_servable_instancecs6g|]tfrtfddjDrqS)c3s,|]}|turjj|jdVqdS).N)objectendpointr> startswith).0baserouter5r6 sz;make_fastapi_class_based_view...)rXany__mro__)r)rrrrr6 s  z1make_fastapi_class_based_view..zOMethods in FastAPI class-based view must have ``self`` as their first argument.)defaultcSsg|] }|jtjjdqS))kind)replaceinspect Parameter KEYWORD_ONLY)r parameterr5r5r6rsN) parameters _serve_clscsg|]}|vr|qSr5r5)rr)routes_to_remover5r6r s)fastapirrfastapi.routingrrroutesremoverr signaturelistrvaluesr{r+r __signature__rrginclude_routerrXrresponse_modelresponse_field outer_type_secure_cloned_response_fieldgetattr) fastapi_apprrrrclass_method_routes new_routerr old_endpoint old_signatureold_parametersold_self_parameternew_self_parameternew_parameters new_signature serve_clsr5)rrrrr6make_fastapi_class_based_viewsZ         "rsockc Csz%|tjtjdttdr|tjtjdWdS|tjddWdStyA}ztd|dWYd}~dSd}~ww) zMutate a socket object to allow multiple process listening on the same port. Returns: success: whether the setting was successful. r SO_REUSEPORTTz'Setting SO_REUSEPORT failed because of z. SO_REUSEPORT is disabled.NF) setsockoptsocket SOL_SOCKET SO_REUSEADDRhasattrrrloggerdebug)rrrr5r5r6set_socket_reuse_ports   rc @s|eZdZdZdeeeffddZedefddZ ede e fdd Z d d Z d ed edede efddZddZdS)ASGIAppReplicaWrapperz;Provides a common wrapper for replicas running an ASGI app. app_or_funccCs@t|r ||_n||_tt|jddddd|_t|j_dS)NonF)lifespan log_level log_config access_log)r isfunction _asgi_apprr_serve_asgi_lifespanr)r4rr5r5r6rc.s   zASGIAppReplicaWrapper.__init__r1cCs|jSr2rr3r5r5r6appEszASGIAppReplicaWrapper.appcCst|jtr |jjSdSr2)rXrrdocs_urlr3r5r5r6 docs_pathIs zASGIAppReplicaWrapper.docs_pathcsfddlm}||jjtjd|jIdH|jjr!tdWddS1s,wYdS)NrLoggingContextlevelz=ASGI lifespan startup failed. Check replica logs for details.) ray.serve._private.logging_utilsrrrloggingWARNINGstartup should_exitrxr4rr5r5r6_run_asgi_lifespan_startupNs "z0ASGIAppReplicaWrapper._run_asgi_lifespan_startupr.r/r0cs||||IdHdS)z Calls into the wrapped ASGI app.Nr)r4r.r/r0r5r5r6ryZs zASGIAppReplicaWrapper.__call__csVddlm}||jjtjd|jIdHWddS1s$wYdS)Nrrr)rrrrrrshutdownrr5r5r6__del__is  "zASGIAppReplicaWrapper.__del__N)r<r=r>rfr rrrcrrr rZrrrrrryr r5r5r5r6r+s$  r middlewarescCsN|durg}t|tstd|D]}tt|ts$tdt|dq|S)a4Validate the return value of HTTP proxy callback. Middlewares should be a list of Starlette middlewares. If it is None, we will treat it as an empty list. If it is not a list, we will raise an error. If it is a list, we will check if all the items in the list are Starlette middlewares. Nz@HTTP proxy callback must return a list of Starlette middlewares.zMHTTP proxy callback must return a list of Starlette middlewares, instead got z type item in the list.)rXr ValueError issubclassrCr)r  middlewarer5r5r6#validate_http_proxy_callback_returnrs  rc@s0eZdZdefddZdededefddZd S) RequestIdMiddlewarercCrsr2)_app)r4rr5r5r6rcruzRequestIdMiddleware.__init__r.r/r0cs\t|d}|tdurt|tdtffdd }||||IdHdS)Nr.rdcsJ|ddkrt|d}|d|ddkr|d<|IdHdS)NrCrUrz X-Request-IDzwebsocket.accept)rrg)rdrW request_idr0r5r6send_with_request_ids    z:RequestIdMiddleware.__call__..send_with_request_id)rrr"r&rgrr)r4r.r/r0rWrr5rr6rys   zRequestIdMiddleware.__call__N) r<r=r>rrcrrrryr5r5r5r6rsrrcCs`ttg|D]&}ttjtdkr|j|fi|j}q|j|g|jRi|j }q|S)zyWrap the ASGI app with the provided middlewares. The built-in RequestIdMiddleware will always be applied first. z0.35.0) rrrparser8 __version__roptionsargskwargs)rr r r5r5r6_apply_middlewaressrF)enable_so_reuseport http_options event_looprcsRt|jtt|jrtjntjtj}|rt|z | |j|j fWnt yB}zt d|jd|j d|d}~wwt jt d_i}|jrz|jrz|j|jd}|jra|j|d<|jri|j|d<td |jd|j d |jtjtjfd d fd |j|j |j|j|ddddd |d}dd |_||j|gdS)z{Start an HTTP server to run the ASGI app. Returns a task that blocks until the server exits (e.g., due to error). zFailed to bind to address 'rS'.Nz uvicorn.error) ssl_keyfile ssl_certfilessl_keyfile_password ssl_ca_certszStarting HTTPS server on z with SSL certificate: csSr2r5r5rr5r6z(start_asgi_http_server..ToffF) factoryhostport root_pathtimeout_keep_alivelooprrrr)configcSsdSr2r5r5r5r5r6r$r%)sockets)rr rrr(AF_INET6AF_INET SOCK_STREAMrbindr)OSErrorrxrCRITICAL getLoggerrrr r!r"rinfouvicornServerrr*keep_alive_timeout_sinstall_signal_handlersrr)rrrrrrr ssl_kwargsserverr5r#r6start_asgi_http_serversj       r=excrequest_timeout_srcCst|trtddd|d|ddSt|tjr+d|d}t|td d|dSt|ttfrKt|t rCtj d |d d id tdd|j dSt|t t fr_tj d |d d id nt dtddt|dS)NTzRequest z timed out after zs.)ris_errorrdzClient for request z" disconnected, cancelling request.izRequest failed: log_to_stderrF)extraz'Request failed due to unexpected error.i)rX TimeoutErrorr$rECancelledErrorrr6r)r*rwarningrdrrrZ)r>r?rrdr5r5r6get_http_response_statuss<      rHrVresponse_startedcCs"|s|jdvr gSt|j|jdS)N)r@rD)rO)rr_rd)rVrIr5r5r6send_http_response_on_exception-s rJcCs@t|}tpddkr t|_|jstr|jpt|_|jpg|_|S)z7Enhanced configuration with component-specific options.r)rrr9r?r!r rr5r5r6$configure_http_options_with_defaults8s   rLcCs6t|}trtdtd|jttt|S)Nz1Calling user-provided callback from import path 'r)rr rr6r extendrr%rKr5r5r6configure_http_middlewaresLsrNre)crErr\rrr collectionsrcopyr dataclassesrtypingrrrrr r r r r r8r7rrfastapi.encodersr packagingrstarlette.datastructuresrstarlette.middlewarerstarlette.typesrrrrruvicorn.configruvicorn.lifespan.onrray._common.network_utilsrray._common.pydantic_compatrray.exceptionsrrray.serve._private.commonrray.serve._private.constantsrr r!r"r#)ray.serve._private.proxy_request_responser$ray.serve._private.utilsr%r&r'ray.serve.configr(ray.serve.exceptionsr)r*r+r5rr-rYrLintr_r`rjrkrrboolrrrrrAbstractEventLoopTaskr=rfloatrZrHrJrLrNr5r5r5r6s   ,               ' \oG  M &