sse-starlette

repository·main·Indexed 21 days ago

A production-ready implementation of Server-Sent Events (SSE) for Starlette and FastAPI. It provides standards-compliant streaming, automatic connection management, and support for modern async patterns via the EventSourceResponse class. Features include structured events with ServerSentEvent, JSON support via JSONServerSentEvent, custom ping messages, memory channels using anyio, and cooperative server shutdown mechanisms.

Tokens: 6.4K
Snippets: 22
Records: 29
Agent score: 72%

What's inside sse-starlette

Overview of sse-starlette examples

main

The repository includes several runnable examples demonstrating different features of sse-starlette:

Example	Feature	Dependencies
`01_basic_sse.py`	Basic streaming (Starlette + FastAPI), conditional data	`fastapi`, `uvicorn`
`02_broadcasting.py`	Multi-client broadcasting via per-client queues	`fastapi`, `uvicorn`
`03_database_streaming.py`	Thread-safe DB sessions in SSE generators	`fastapi`, `uvicorn`, `sqlalchemy`, `aiosqlite`
`04_advanced_features.py`	Custom ping, error handling, separators, headers	`fastapi`, `uvicorn`
`05_memory_channels.py`	Memory channels with `data_sender_callable`	`uvicorn`
`06_send_timeout.py`	Frozen client detection via `send_timeout`	`uvicorn`
`07_cooperative_shutdown.py`	Graceful shutdown with farewell events (v3.3.0)	`uvicorn`

Handle cooperative server shutdown

main

By default, generators receive a CancelledError immediately when the server shuts down. To allow your generator to send 'farewell' events or perform cleanup, use the shutdown_event and shutdown_grace_period parameters.

The library sets the provided shutdown_event when shutdown begins.
Your generator checks this event and yields final messages.
If the generator exits within shutdown_grace_period, it shuts down cleanly.

Note: shutdown_grace_period must be less than your ASGI server's graceful shutdown timeout.

import anyio
from sse_starlette import EventSourceResponse

async def graceful_stream(request):
    shutdown_event = anyio.Event()

    async def generate():
        try:
            while not shutdown_event.is_set():
                yield {"data": "tick"}
                with anyio.move_on_after(1.0):
                    await shutdown_event.wait()
            # Shutdown detected — send farewell event
            yield {"event": "shutdown", "data": "Server is shutting down"}
        except anyio.get_cancelled_exc_class():
            # Grace period expired
            raise

    return EventSourceResponse(
        generate(),
        shutdown_event=shutdown_event,
        shutdown_grace_period=5.0,
    )

How shutdown and cancellation works in EventSourceResponse

main

The EventSourceResponse uses an anyio task group to manage the lifecycle of an SSE connection. It runs four concurrent tasks in a 'race' pattern using a cancel_on_finish wrapper. Whichever task completes first triggers a cancellation of all other sibling tasks.

The four concurrent tasks are:

_stream_response: Pushes SSE data from your generator to the client.
_ping: Sends keepalive pings approximately every 15 seconds.
_listen_for_exit_signal_with_grace: Monitors for server shutdown signals.
_listen_for_disconnect: Monitors for client-side connection closures.

Key behaviors based on which task wins the race:

Scenario	Winning Task	Resulting Behavior
Normal Completion	`_stream_response`	The generator exhausts naturally. The connection exits cleanly, often allowing a 'farewell' event to be sent.
Client Disconnect	`_listen_for_disconnect`	The client closes the connection. The generator receives a `CancelledError`.
Server Shutdown (No Grace)	`_listen_for_exit_signal_with_grace`	The server receives a SIGTERM/SIGINT. The generator receives a `CancelledError` immediately.
Server Shutdown (With Grace)	`_stream_response`	If you provide a `shutdown_event` and `shutdown_grace_period`, the generator can detect the shutdown signal, yield a final 'farewell' event, and exit cleanly.
Server Shutdown (Grace Timeout)	`_listen_for_exit_signal_with_grace`	If the generator ignores the shutdown signal and exceeds the `shutdown_grace_period`, the task group forces a cancellation via `CancelledError`.
Send Timeout	`_stream_response`	If a `send()` operation hangs and exceeds its timeout, a `SendTimeoutError` is raised, cancelling all other tasks.

Detect client disconnection

main

To prevent resource leaks and hanging connections, always check if the client has disconnected within your streaming loop using await request.is_disconnected().

async def monitored_stream(request):
    events_sent = 0
    try:
        while events_sent < 100:
            if await request.is_disconnected():
                print(f"Client disconnected after {events_sent} events")
                break
            
            yield {"data": f"Event {events_sent}"}
            events_sent += 1
            await asyncio.sleep(1)
            
    except asyncio.CancelledError:
        print("Stream cancelled")
        raise

Quick Start with SSE in Starlette

main

To implement a basic Server-Sent Events endpoint, create an asynchronous generator that yields dictionaries containing the data key, and wrap it in an EventSourceResponse.

import asyncio
from starlette.applications import Starlette
from starlette.routing import Route
from sse_starlette import EventSourceResponse

async def generate_events():
    for i in range(10):
        yield {"data": f"Event {i}"}
        await asyncio.sleep(1)

async def sse_endpoint(request):
    return EventSourceResponse(generate_events())

app = Starlette(routes=[Route("/events", sse_endpoint)])

Implement cooperative shutdown with shutdown_event and shutdown_grace_period
main
To prevent abrupt connection closures during server shutdown, you can implement a cooperative shutdown pattern. By providing a shutdown_event (an anyio.Event) and a shutdown_grace_period (in seconds) to the EventSourceResponse, your generator can listen for the shutdown signal and perform cleanup or send a final 'farewell' message before the connection is forcibly closed.
The workflow is:
1. The server receives a shutdown signal (SIGTERM/SIGINT).
2. The _listen_for_exit_signal_with_grace task detects this and sets the provided shutdown_event.
3. Your generator, which should be checking this event, yields a final event.
4. Once your generator finishes, the _stream_response task completes, and the task group exits cleanly.
https://github.com/sysid/sse-starlette/blob/main/ARCHITECTURE.md
Install sse-starlette and example dependencies
main
To use the core library, install via pip:
```
pip install sse-starlette
```
To run the provided examples, you should install the optional dependency groups:
- Standard examples: Installs fastapi, uvicorn, and pydantic.
- Database examples: Installs sqlalchemy and aiosqlite in addition to the standard examples.
Use the following commands:
```
# For most examples
pip install 'sse-starlette[examples]'

# For Example 03 (database streaming)
pip install 'sse-starlette[examples-db]'
```
```
pip install 'sse-starlette[examples]'
pip install 'sse-starlette[examples-db]'
```
https://github.com/sysid/sse-starlette/blob/main/examples/README.md

Install sse-starlette

main

You can install sse-starlette using pip or uv. Depending on your needs, you can also install optional extras for examples, database support, or specific ASGI servers.

# Basic installation
pip install sse-starlette
uv add sse-starlette

# Install with examples (includes fastapi, uvicorn, pydantic)
uv add sse-starlette[examples]

# Install with database extras (includes sqlalchemy, aiosqlite)
uv add sse-starlette[examples,examples-db]

# Install with recommended ASGI servers
uv add sse-starlette[uvicorn,granian,daphne]

pip install sse-starlette

Run sse-starlette examples
main
You can run the examples using uv (which handles the inline PEP 723 dependency declarations) or by manually installing dependencies and using Python.
Using uv (Recommended)
```
uv run examples/01_basic_sse.py
```
Using pip and Python
```
pip install sse-starlette fastapi uvicorn
python examples/01_basic_sse.py
```
Testing the stream
Once the server is running, use curl with the -N (no-buffer) flag to view the stream in your terminal:
```
curl -N http://localhost:8000/<endpoint>
```
Note: Check the specific example file's docstring to find the correct <endpoint> and curl command.
```
curl -N http://localhost:8000/<endpoint>
```
https://github.com/sysid/sse-starlette/blob/main/examples/README.md
Implement graceful shutdown with shutdown_event and shutdown_grace_period
main
To prevent abrupt termination of SSE streams during server shutdown, you can use cooperative shutdown.
1. Pass an anyio.Event to the shutdown_event parameter.
2. Pass a shutdown_grace_period (in seconds) to allow the generator time to finish.
3. In your generator, watch the shutdown_event. When it is set, you can send 'farewell' messages before exiting.
Warning: The shutdown_grace_period should be less than your ASGI server's (e.g., Uvicorn's) graceful shutdown timeout to ensure the process isn't killed before the period expires.
```
import anyio
from sse_starlette.sse import EventSourceResponse

async def event_generator(shutdown_event: anyio.Event):
    try:
        for i in range(100):
            if shutdown_event.is_set():
                yield {"data": "Goodbye!"}
                return
            yield {"data": f"msg {i}"}
            await asyncio.sleep(1)
    except Exception:
        pass

@app.get("/stream")
async def stream():
    shutdown_evt = anyio.Event()
    return EventSourceResponse(
        event_generator(shutdown_evt), 
        shutdown_event=shutdown_evt,
        shutdown_grace_period=5.0
    )
```
https://github.com/sysid/sse-starlette/blob/main/sse_starlette/sse.py
Avoid buffering issues with CDNs and Load Balancers
main
Be aware of the following network infrastructure behaviors that can break real-time SSE delivery:
- Cloudflare: Buffers approximately 100KB before flushing to clients.
- Akamai: Edge servers buffer by default.
- F5 Load Balancers: Buffer responses by default.
https://github.com/sysid/sse-starlette/blob/main/README.md
Configure HAProxy timeouts for SSE heartbeats
main
When using HAProxy, ensure that your timeout client and timeout server settings are greater than the frequency of your SSE heartbeat (ping) interval to prevent the connection from being closed prematurely.
```
# Ensure timeouts > ping interval
timeout client 60s    # If ping every 45s
timeout server 60s
```
https://github.com/sysid/sse-starlette/blob/main/README.md

sse-starlette

What's inside sse-starlette

Using uv (Recommended)

Using pip and Python

Testing the stream