o jRL@sdZddlZddlZddlZddlZddlmZmZddlmZm Z ddl m Z m Z m Z mZmZmZmZmZmZmZddlZddlmZddlmZddlmZdd lmZdd lmZm Z m!Z!m"Z"dd l#m$Z$dd l%m&Z&m'Z'e(e)Z*eGd ddZ+e,Z-de+fddZ.ddZ/d#ddZ0d#ddZ1de!de!fddZ2Gddde3Z4GdddZ5zddl6m7Z7e7j8e5_9e5j8e7_8Wne:ye*;d Ynweee&e fZ?ee?Z@e e?ZAeeAe@fZBGd!d"d"eZCdS)$uServer-Sent Events response for Starlette / FastAPI. Intentional divergence from ``starlette.responses.StreamingResponse`` -------------------------------------------------------------------- ``EventSourceResponse`` is modelled on Starlette's ``StreamingResponse`` and re-syncs most of its behaviour (WebSocket denial, ``collapse_excgroups()`` around the task group, ``memoryview`` chunk handling). The following points are deliberate divergences — DO NOT "fix" them without reading the rationale: 1. ASGI ``spec_version >= 2.4`` fast path is NOT adopted. Upstream short-circuits to ``await stream_response(send)`` and converts ``OSError`` into ``ClientDisconnect``, skipping ``listen_for_disconnect``. We keep ``_listen_for_disconnect`` running because it (a) invokes ``client_close_handler_callable`` on disconnect, (b) flips ``self.active = False`` so ``_ping`` and the cooperative shutdown grace loop exit promptly. Adopting the upstream fast path would regress both features. 2. ``_wrap_websocket_denial_send`` is inlined in this module rather than inherited from ``starlette.responses.Response``. The helper landed on Starlette ``main`` after our minimum pin (``starlette>=0.41.3``); inline until the floor moves past the release that contains it. 3. ``collapse_excgroups()`` is vendored in ``sse_starlette._utils`` rather than imported from ``starlette._utils`` (private module). N) dataclassfield)datetimetimezone) Any AsyncIterable AwaitableCallable CoroutineIteratorMappingOptionalSetUnion)BackgroundTask)iterate_in_threadpool)MutableHeaders)Response)ReceiveScopeSendMessage)collapse_excgroups)ServerSentEvent ensure_bytesc@s6eZdZUdZeedZeej e d<dZ e e d<dS)_ShutdownStatezPer-thread state for shutdown coordination. Issue #152 fix: Uses threading.local() instead of ContextVar to ensure one watcher per thread rather than one per async context. )default_factoryeventsFwatcher_startedN) __name__ __module__ __qualname____doc__rsetrranyioEvent__annotations__rboolr(r(D/home/kuhnn/.local/lib/python3.10/site-packages/sse_starlette/sse.pyr>s rreturncCs$ttdd}|durt}|t_|S)z4Get or create shutdown state for the current thread.shutdown_stateN)getattr _thread_staterr+)stater(r(r)_get_shutdown_stateNs r/cCsNzttj}t|dr|j}t|dr|WSWdSWdSty&YdSw)am Try to get uvicorn Server instance via signal handler introspection. When uvicorn registers signal handlers, they're bound methods on the Server instance. We can retrieve the Server from the handler's __self__ attribute. Returns None if: - Not running under uvicorn - Signal handler isn't a bound method - Any introspection fails __self__ should_exitN)signal getsignalSIGTERMhasattrr0 Exception)handlerserverr(r(r)_get_uvicorn_serverWs    r9cstt}t}z. tjr ntjr|dur|jrdt_n tdIdHq t|jD]}| q)Wd|_ dSd|_ w)ag Poll for shutdown and broadcast to all events in this context. One watcher runs per thread (event loop). Checks two shutdown sources: 1. AppStatus.should_exit - set when our monkey-patch works 2. uvicorn Server.should_exit - via signal handler introspection (Issue #132 fix) When either becomes True, signals all registered events. TNg?F) r/r9 AppStatusr1enable_automatic_graceful_drainr$sleeplistrr#r)r.uvicorn_servereventr(r(r)_shutdown_watcherns(  r@cCsLt}|js$d|_z t}|tWdSty#d|_YdSwdS)zDEnsure the shutdown watcher is running for this thread (event loop).TFN)r/rasyncioget_running_loop create_taskr@ RuntimeError)r.loopr(r(r)$_ensure_watcher_started_on_this_loops  rFsendcsdtddffdd }|S)a;Mirror of ``starlette.responses.Response._wrap_websocket_denial_send``. Divergence #2 (see module docstring): inlined because the helper landed on Starlette ``main`` (commit 9ee9519) after our minimum pin ``starlette>=0.41.3``. Drop this once the floor moves past the release that contains it. messager*Ncs8|d}|dvri|dd|i}|IdHdS)Ntype>http.response.bodyhttp.response.startz websocket.r()rH message_typerGr(r)wrappeds z,_wrap_websocket_denial_send..wrapped)r)rGrNr(rMr)_wrap_websocket_denial_sends rOc@s eZdZdS)SendTimeoutErrorN)rr r!r(r(r(r)rPsrPc@sNeZdZUdZdZdZdZeee d<e ddZ e dd Z e d d Z dS) r:z\Helper to capture a shutdown signal from Uvicorn so we can gracefully terminate SSE streams.FTNoriginal_handlercC dt_dS)aJ Prevent automatic SSE stream termination on server shutdown. WARNING: When disabled, you MUST set AppStatus.should_exit = True at some point during shutdown, or streams will never close and the server will hang indefinitely (or until uvicorn's graceful shutdown timeout expires). FNr:r;r(r(r(r) disable_automatic_graceful_drains z*AppStatus.disable_automatic_graceful_draincCrR)a Re-enable automatic SSE stream termination on server shutdown. This restores the default behavior where SIGTERM triggers immediate stream draining. Call this to undo a previous call to disable_automatic_graceful_drain(). TNrSr(r(r(r)$enable_automatic_graceful_drain_modes z.AppStatus.enable_automatic_graceful_drain_modecOs.tjrdt_tjdurtj|i|dSdS)NT)r:r;r1rQ)argskwargsr(r(r) handle_exits  zAppStatus.handle_exit)rr r!r"r1r;rQr r r& staticmethodrTrUrXr(r(r(r)r:s    r:)ServerzHUvicorn not installed. Graceful shutdown on server termination disabled.c@szeZdZdZdZdZ            d0ded ed ee e e fd e d ee d eedee dee ge fdee gedfdeedee egedfdeejdeddfddZedeeeffddZejdeeefddfddZd1deddfdd Zd!eddfd"d#Zd$eddfd%d&Zed2d'd(Z d2d)d*Z!d!eddfd+d,Z"d-e#d$ed!eddfd.d/Z$dS)3EventSourceResponseagStreaming response implementing the SSE (Server-Sent Events) specification. Args: content: Async iterable or sync iterator yielding SSE event data. status_code: HTTP status code. Default: 200. headers: Additional HTTP headers. media_type: Response media type. Default: "text/event-stream". background: Background task to run after response completes. ping: Ping interval in seconds (0 to disable). Default: 15. sep: Line separator for SSE messages ("\r\n", "\r", or "\n"). ping_message_factory: Callable returning custom ping ServerSentEvent. data_sender_callable: Async callable for push-based data sending. send_timeout: Timeout in seconds for individual send operations. client_close_handler_callable: Async callback on client disconnect. shutdown_event: Optional ``anyio.Event`` set by the library when server shutdown is detected. Generators can watch this event to send farewell messages and exit cooperatively instead of receiving CancelledError. shutdown_grace_period: Seconds to wait after setting ``shutdown_event`` before force-cancelling the generator. Must be >= 0. Should be less than your ASGI server's graceful shutdown timeout. Default: 0 (immediate cancel, identical to pre-v3.3.0 behavior).  Ntext/event-streamrcontent status_codeheaders media_type backgroundpingsepping_message_factorydata_sender_callable)NNN send_timeoutclient_close_handler_callableshutdown_eventshutdown_grace_periodr*cCs|dvr td||p|j|_t|tr||_nt||_||_|dur)|jn||_||_ | |_ | |_ t }|durA| ||ddd|d<d|d<|||dur[|jn||_||_| |_| d krltd | |_| |_d |_t|_dS) N)Nr]  z'sep must be one of: \r\n, \r, \n, got: z Cache-Controlzno-storez keep-alive ConnectionnozX-Accel-Bufferingrz"shutdown_grace_period must be >= 0T) ValueErrorDEFAULT_SEPARATORrf isinstancer body_iteratorrrarcrdrhrirupdate setdefault init_headersDEFAULT_PING_INTERVAL ping_intervalrgrj_shutdown_event_shutdown_grace_periodactiver$Lock _send_lock)selfr`rarbrcrdrerfrgrhrirjrkrl_headersr(r(r)__init__s6      zEventSourceResponse.__init__cCs|jSN)_ping_intervalrr(r(r)ryOsz!EventSourceResponse.ping_intervalvaluecCs0t|ttfs td|dkrtd||_dS)Nzping interval must be intrz$ping interval must be greater than 0)rsintfloat TypeErrorrqr)rrr(r(r)rySs  FforcecCstd)Nz-Compression is not supported for SSE streams.)NotImplementedError)rrr(r(r)enable_compression[sz&EventSourceResponse.enable_compressionrGc s|d|j|jdIdH|j2zL3dHW}t||j}td|t|j }|d|ddIdHWdn1s?wY|r]|j r]t |jdd}|durZ|IdHt q6|j 4IdHd |_|dd d dIdHWdIdHdS1IdHswYdS) zHSend out SSE data to the client as it becomes available in the iterator.rK)rIstatusrbNz chunk: %srJTrIbody more_bodyacloseF)ra raw_headersrtrrfloggerdebugr$ move_on_afterri cancel_calledr,rPr~r|)rrGdatachunk cancel_scoperr(r(r)_stream_response^s4       .z$EventSourceResponse._stream_responsereceivecsX|jr*|IdH}|ddkr%d|_td|jr#||IdHdS|jsdSdS)uqWatch for a disconnect message from the client. Divergence #1 (see module docstring): kept unconditionally instead of adopting Starlette's ASGI 2.4 ``OSError → ClientDisconnect`` fast path, because this loop drives ``client_close_handler_callable`` and flips ``self.active = False`` for ``_ping`` and the shutdown grace loop. NrIzhttp.disconnectFz+Got event: http.disconnect. Stop streaming.)r|rrrj)rrrHr(r(r)_listen_for_disconnectzs   z*EventSourceResponse._listen_for_disconnectc sttjrdStt}t}|j|ztjr#W|j|dS| IdHW|j|dS|j|w)z0Wait for shutdown signal via the shared watcher.N) r:r1rFr/r$r%radddiscardwait)r.r?r(r(r)_listen_for_exit_signals z+EventSourceResponse._listen_for_exit_signalcs|IdH|jr|j|jdkrDt|j|jr2tdIdH|jsWddSWddS1s=wYdSdS)a~Wait for shutdown signal, then optionally give generator a grace period. Issue #167: When a shutdown_event is provided, the library sets it before returning, giving the generator a chance to send farewell events and exit cooperatively. The shutdown_grace_period controls how long to wait before force-cancelling via task group cancellation. Nrg?)rrzr#r{r$rr|r<rr(r(r)"_listen_for_exit_signal_with_graces  "z6EventSourceResponse._listen_for_exit_signal_with_gracec s|jr^t|jIdH|jr|n tdttj |j d}t ||j }t d||j4IdH|jrD|d|ddIdHWdIdHn 1IdHsTwY|jsdSdS)zPeriodically send ping messages to keep the connection alive on proxies. - frequenccy ca every 15 seconds. - Alternatively one can send periodically a comment line (one starting with a ':' character) Nzping - )commentrfzping: %srJTr)r|r$r<rrgrrnowrutcrfrrrr~)rrGsse_ping ping_bytesr(r(r)_pings,    (zEventSourceResponse._pingscopec s|ddkr ttft4IdHKdtgtdfffdd }|fdd|fd d|jjrMj|fd dWdIdHn 1IdHshwYWdn1swwYj dur IdHdSdS) aEntrypoint for Starlette's ASGI contract. We spin up tasks: - _stream_response to push events - _ping to keep the connection alive - _listen_for_exit_signal to respond to server shutdown - _listen_for_disconnect to respond to client disconnect rI websocketNcorocs|IdHjdSr)rcancel)r) task_groupr(r)cancel_on_finishs z6EventSourceResponse.__call__..cancel_on_finishc Sr)rr(rrGr(r) z.EventSourceResponse.__call__..crr)rr(rr(r)rrcs Sr)rr()rrr(r)rr) rOrr$create_task_groupr r start_soonrrhrd)rrrrGrr()rrrGrr)__call__s0  ( zEventSourceResponse.__call__) r^Nr_NNNNNNNNr)Fr*N)%rr r!r"rxrr ContentStreamrr r strrr rr rrrr$r%rpropertyrrysetterr'rrrrrrYrrrrrr(r(r(r)r[sz    I  r[r)Dr"rAloggingr2 threading dataclassesrrrrtypingrrrr r r r r rrr$starlette.backgroundrstarlette.concurrencyrstarlette.datastructuresrstarlette.responsesrstarlette.typesrrrrsse_starlette._utilsrsse_starlette.eventrr getLoggerrrrlocalr-r/r9r@rFrO TimeoutErrorrPr: uvicorn.mainrZrXrQ ImportErrorrrbytesdictContentSyncContentStreamAsyncContentStreamrr[r(r(r(r)sR0         $ &