### Run Example with Python

Source: https://github.com/sysid/sse-starlette/blob/main/examples/README.md

Install dependencies manually and run the basic SSE example using Python. This method requires explicit installation of 'sse-starlette', 'fastapi', and 'uvicorn'.

```bash
pip install sse-starlette fastapi uvicorn
python examples/01_basic_sse.py
```

--------------------------------

### Run Example with UV

Source: https://github.com/sysid/sse-starlette/blob/main/examples/README.md

Use the 'uv' command to run the basic SSE example. Ensure 'sse-starlette', 'fastapi', and 'uvicorn' are installed.

```bash
uv run examples/01_basic_sse.py
```

--------------------------------

### Install sse-starlette and dependencies

Source: https://github.com/sysid/sse-starlette/blob/main/README.md

Install the library using pip. Additional extras are available for examples, database support, and recommended ASGI servers.

```bash
pip install sse-starlette
uv add sse-starlette

# To run the examples (fastapi, uvicorn, pydantic)
uv add sse-starlette[examples]

# Example 03 also needs the DB extras (sqlalchemy, aiosqlite)
uv add sse-starlette[examples,examples-db]

# Recommended ASGI server
uv add sse-starlette[uvicorn,granian,daphne]
```

--------------------------------

### Run Demonstration Script

Source: https://github.com/sysid/sse-starlette/blob/main/README.md

Execute a demonstration script from the examples directory. Ensure you are in the project root.

```bash
python examples/demonstrations/basic_patterns/client_disconnect.py
```

```bash
python examples/demonstrations/production_scenarios/load_simulations.py
```

```bash
python examples/demonstrations/advanced_patterns/error_recovery.py
```

--------------------------------

### JavaScript Client Example

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/README.md

A basic JavaScript example to connect to an SSE endpoint using the native `EventSource` API.

```javascript
const eventSource = new EventSource('/stream');

eventSource.onmessage = function(event) {
  console.log('Received message:', event.data);
};

eventSource.onerror = function(err) {
  console.error('EventSource failed:', err);
  eventSource.close();
};
```

--------------------------------

### Minimal EventSourceResponse Setup

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/07-module-structure.md

Demonstrates the most basic way to set up an endpoint that streams Server-Sent Events using EventSourceResponse.

```python
from sse_starlette import EventSourceResponse

@app.get("/events")
async def stream():
    return EventSourceResponse(generate_events())
```

--------------------------------

### Example Data Sender Callable

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/06-architecture-patterns.md

This example shows how to implement a custom `data_sender_callable` to push data into the stream. It uses an asyncio Queue to hold items that will be processed by the SSE response.

```python
async def push_data(queue: asyncio.Queue):
    for i in range(10):
        await queue.put({"data": f"Item {i}"})
        await asyncio.sleep(1)

response = EventSourceResponse(
    queue,
    data_sender_callable=lambda: push_data(queue)
)
```

--------------------------------

### Testing Snippet: Mock Client

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/README.md

A simple mock client to test SSE endpoints. This example uses `httpx` to connect and receive events.

```python
import asyncio
import httpx

async def test_sse_client(url):
    async with httpx.AsyncClient() as client:
        async with client.stream('GET', url) as response:
            if response.status_code == 200:
                async for chunk in response.aiter_bytes():
                    print(chunk.decode(), end='')
            else:
                print(f"Error: {response.status_code}")

# Example usage:
# asyncio.run(test_sse_client('http://localhost:8000/stream'))
```

--------------------------------

### Connection Limiter Example

Source: https://github.com/sysid/sse-starlette/blob/main/README.md

Implement a connection limiter using asyncio.Semaphore to control the maximum number of concurrent connections to an SSE endpoint.

```python
# Connection limits
class ConnectionLimiter:
    def __init__(self, max_connections=100):
        self.semaphore = asyncio.Semaphore(max_connections)
    
    async def limited_endpoint(self, request):
        async with self.semaphore:
            return EventSourceResponse(generate_events())
```

--------------------------------

### ASGI Server Configuration: Uvicorn

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/README.md

Configure Uvicorn for running Starlette/FastAPI applications with SSE support. This example shows basic Uvicorn command-line arguments.

```bash
uvicorn main:app --reload --host 0.0.0.0 --port 8000
```

--------------------------------

### Basic SSE Streaming with FastAPI

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/07-module-structure.md

Use this pattern for simple data streaming where each yielded item is treated as a data field. Ensure FastAPI and sse-starlette are installed.

```python
from fastapi import FastAPI
from sse_starlette import EventSourceResponse

app = FastAPI()

async def generate():
    yield {"data": "item"}

@app.get("/events")
async def events():
    return EventSourceResponse(generate())
```

--------------------------------

### Memory Channel Setup for Pub/Sub

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/05-configuration.md

Utilize anyio's memory object streams to create a publish-subscribe mechanism for SSE. A producer function sends messages to a channel, which is then consumed by EventSourceResponse.

```python
import anyio
from functools import partial

async def broadcast_producer(send_channel, messages: List[str]):
    async with send_channel:
        for msg in messages:
            await send_channel.send({"data": msg})

@app.get("/events")
async def broadcast_events():
    send_ch, recv_ch = anyio.create_memory_object_stream(maxsize=100)
    messages = ["msg1", "msg2", "msg3"]
    
    return EventSourceResponse(
        recv_ch,
        data_sender_callable=partial(broadcast_producer, send_ch, messages)
    )
```

--------------------------------

### Minimal SSE Endpoint (FastAPI)

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/README.md

A basic Server-Sent Events endpoint using FastAPI. Ensure you have FastAPI and uvicorn installed.

```python
from fastapi import FastAPI
from sse_starlette.sse import EventSourceResponse

app = FastAPI()

async def custom_generator():
    for i in range(10):
        yield f"data: {i}\n\n"

@app.get('/stream')
async def stream():
    return EventSourceResponse(custom_generator())
```

--------------------------------

### Create Basic ServerSentEvent

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/02-serversentevent.md

Demonstrates creating a simple `ServerSentEvent` with just data, and a more complex event including event type, ID, and retry interval. The `.encode()` method is used to get the SSE-formatted bytes.

```python
from sse_starlette import ServerSentEvent

# Simple message
event = ServerSentEvent(data="Hello, client!")
print(event.encode())
# Output: b'data: Hello, client!\r\n\r\n'
```

```python
# Full-featured event
event = ServerSentEvent(
    data="Order #123 shipped",
    event="order",
    id="123",
    retry=5000
)
print(event.encode())
# Output: b'id: 123\r\nevent: order\r\ndata: Order #123 shipped\r\nretry: 5000\r\n\r\n'
```

--------------------------------

### Multi-threaded Deployment with Independent Event Loops

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/06-architecture-patterns.md

This example shows how to deploy an SSE application in a multi-threaded environment using `ThreadPoolExecutor`. Each thread runs the SSE app with its own independent asyncio event loop, ensuring proper isolation and preventing context variable confusion.

```python
import threading
from concurrent.futures import ThreadPoolExecutor

executor = ThreadPoolExecutor(max_workers=4)

for i in range(4):
    executor.submit(run_sse_app)

def run_sse_app():
    asyncio.set_event_loop(asyncio.new_event_loop())
    asyncio.run(serve())
```

--------------------------------

### Production Setup with Graceful Shutdown

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/05-configuration.md

Configure EventSourceResponse for production with graceful shutdown. Use a long ping interval and send timeout for efficiency and stability. The shutdown_grace_period allows a brief window for cleanup.

```python
import anyio
from fastapi import FastAPI
from sse_starlette import EventSourceResponse

app = FastAPI()

async def generate():
    shutdown_event = anyio.Event()
    
    async def stream():
        try:
            while not shutdown_event.is_is_set():
                yield {"data": "update"}
                with anyio.move_on_after(5.0):
                    await shutdown_event.wait()
            yield {"event": "shutdown", "data": "Server closing"}
        except anyio.get_cancelled_exc_class():
            raise
    
    return EventSourceResponse(
        stream(),
        ping=30,  # Long ping to reduce overhead
        send_timeout=60,  # Long timeout for slow clients
        shutdown_event=shutdown_event,
        shutdown_grace_period=5.0,
        headers={
            "Cache-Control": "no-store",
            "X-Content-Type-Options": "nosniff"
        }
    )

@app.get("/events")
async def events():
    return await generate()
```

```bash
uvicorn app:app --timeout-graceful-shutdown 10
```

--------------------------------

### Graceful Shutdown with Farewell Events

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/03-appstatus-shutdown.md

This example shows how to implement a stream that sends a farewell message before closing upon detecting a shutdown signal. It utilizes `anyio.Event` for signaling and specifies a shutdown grace period.

```python
import anyio
from sse_starlette import EventSourceResponse

async def generate_with_grace():
    shutdown_event = anyio.Event()
    
    async def stream():
        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
            yield {"event": "goodbye", "data": "Server shutting down"}
        except anyio.get_cancelled_exc_class():
            raise
    
    return EventSourceResponse(
        stream(),
        shutdown_event=shutdown_event,
        shutdown_grace_period=5.0
    )

```

--------------------------------

### Database Streaming Pattern

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/README.md

Stream data directly from a database to SSE clients. This example assumes a hypothetical async database cursor.

```python
import asyncio
from sse_starlette.sse import EventSourceResponse

async def fake_db_cursor():
    # Simulate fetching data from a database
    for i in range(5):
        await asyncio.sleep(0.5)
        yield {"record_id": i, "data": f"Data row {i}"}

async def db_stream_generator():
    async for record in fake_db_cursor():
        yield f"data: {record}\n\n"

response = EventSourceResponse(db_stream_generator())
```

--------------------------------

### CDN-Friendly Setup with Proxy Caching

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/05-configuration.md

Configure EventSourceResponse for CDN caching by setting appropriate Cache-Control and Vary headers. A shorter ping interval is used for better connection stability behind proxies.

```python
response = EventSourceResponse(
    stream(),
    ping=15,  # Frequent pings for connection stability
    headers={
        "Cache-Control": "public, max-age=29",  # Allow CDN caching
        "Vary": "Accept-Encoding"  # Inform CDN of compression
    }
)
```

--------------------------------

### JavaScript SSE Client Example

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/08-quick-reference.md

Connect to an SSE endpoint using the EventSource API in JavaScript. Handles 'message' and custom events, and includes error handling to close the connection on failure.

```javascript
const eventSource = new EventSource('/events');

eventSource.addEventListener('message', (event) => {
    console.log('Data:', event.data);
});

eventSource.addEventListener('custom-event', (event) => {
    console.log('Custom:', event.data);
});

eventSource.onerror = (event) => {
    console.error('Error:', event);
    eventSource.close();
};
```

--------------------------------

### Get SSE Starlette Version

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/07-module-structure.md

Retrieves the installed version of the sse-starlette library. This is useful for checking compatibility or reporting issues.

```python
import sse_starlette
print(sse_starlette.__version__)  # "3.4.5"
```

--------------------------------

### Minimal SSE Endpoint (Starlette)

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/README.md

A basic Server-Sent Events endpoint using Starlette. Ensure you have Starlette and uvicorn installed.

```python
from starlette.applications import Starlette
from starlette.responses import Response
from sse_starlette.sse import EventSourceResponse

async def custom_generator():
    for i in range(10):
        yield f"data: {i}\n\n"

app = Starlette(routes=[
    Starlette.route('/stream', endpoint=lambda r: EventSourceResponse(custom_generator()))
])
```

--------------------------------

### Error Handling and Recovery

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/README.md

Implement error handling within your SSE generator. This example shows how to catch exceptions and send an error event.

```python
import asyncio
from sse_starlette.sse import EventSourceResponse, ServerSentEvent

async def error_handling_generator():
    try:
        for i in range(5):
            yield f"data: Processing item {i}\n\n"
            await asyncio.sleep(0.5)
            if i == 3:
                raise ValueError("Something went wrong!")
    except Exception as e:
        yield ServerSentEvent(data=f"ERROR: {e}", event="error")

response = EventSourceResponse(error_handling_generator())
```

--------------------------------

### Database Connection Cleanup on Disconnect

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/05-configuration.md

Example of using the client_close_handler_callable to clean up database sessions when a client disconnects from the SSE stream.

```python
async def cleanup_on_disconnect(message):
    # This is called from _listen_for_disconnect
    # You can log, update stats, etc.
    logger.info("Client disconnected, cleaning up database session")

@app.get("/events")
async def stream_with_db():
    return EventSourceResponse(
        stream_from_db(),
        client_close_handler_callable=cleanup_on_disconnect
    )
```

--------------------------------

### Setting Shutdown Grace Period

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/05-configuration.md

This example shows how to configure a grace period for the generator to finish sending messages after the shutdown event is triggered. Ensure this period is less than the ASGI server's shutdown timeout.

```python
response = EventSourceResponse(
    stream(),
    shutdown_event=shutdown_event,
    shutdown_grace_period=3.0  # Wait 3 seconds for generator to finish
)
```

--------------------------------

### Queue/Channel-Based Broadcasting for SSE

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/06-architecture-patterns.md

This pattern is suitable for broadcasting messages to multiple consumers simultaneously. Each client gets its own queue, and a producer can broadcast messages to all connected clients, with individual queue management for slow clients.

```python
class BroadcastStream:
    def __init__(self, request: Request, broadcaster):
        self.queue = broadcaster.add_client()
    
    def __aiter__(self):
        return self
    
    async def __anext__(self):
        if await self.request.is_disconnected():
            raise StopAsyncIteration
        return await self.queue.get()

@app.get("/events")
async def events(request: Request):
    stream = broadcaster.create_stream(request)
    return EventSourceResponse(stream)

@app.post("/broadcast")
async def broadcast(message: str):
    await broadcaster.broadcast(message)
```

--------------------------------

### Generator with Shutdown Event Handling

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/05-configuration.md

This example demonstrates how a generator can utilize `shutdown_event` to send farewell messages before the server shuts down. It waits for the event to be set, sends a final message, and then exits cleanly.

```python
import anyio

async def graceful_stream():
    shutdown_event = anyio.Event()
    
    async def generate():
        try:
            while not shutdown_event.is_is_set():
                yield {"data": "tick"}
                with anyio.move_on_after(1.0):
                    await shutdown_event.wait()
            # Shutdown detected
            yield {"event": "goodbye", "data": "Server shutting down"}
        except anyio.get_cancelled_exc_class():
            raise
    
    return EventSourceResponse(
        generate(),
        shutdown_event=shutdown_event,
        shutdown_grace_period=5.0
    )
```

--------------------------------

### SSE Configuration: Timeout Protection

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/README.md

Implement timeout protection for Server-Sent Events to prevent indefinite waits. This example sets a send timeout of 10 seconds.

```python
from sse_starlette.sse import EventSourceResponse

async def custom_generator():
    for i in range(10):
        yield f"data: {i}\n\n"

# Set send_timeout to 10 seconds
response = EventSourceResponse(custom_generator(), send_timeout=10000)
```

--------------------------------

### Create JSONServerSentEvent

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/02-serversentevent.md

Shows how to create a `JSONServerSentEvent` where the data is a Python dictionary. The dictionary is automatically JSON-encoded and sent as the event data. This example includes event type and ID.

```python
from sse_starlette import JSONServerSentEvent

event = JSONServerSentEvent(
    data={"user_id": 42, "action": "login", "timestamp": "2024-01-15T10:30:00Z"},
    event="auth",
    id="evt-001"
)
print(event.encode())
# Output: b'id: evt-001\r\nevent: auth\r\ndata: {"user_id":42,"action":"login","timestamp":"2024-01-15T10:30:00Z"}\r\n\r\n'
```

--------------------------------

### ServerSentEvent with Custom Line Separators

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/02-serversentevent.md

Shows how to specify custom line separators for SSE events using the `sep` parameter. Examples include Unix (LF) and old Mac (CR) line endings.

```python
# Unix line endings (LF only)
event = ServerSentEvent(data="test", sep="\n")
print(event.encode())
# Output: b'data: test\n\n'
```

```python
# Old Mac line endings (CR only)
event = ServerSentEvent(data="test", sep="\r")
print(event.encode())
# Output: b'data: test\r\r'
```

--------------------------------

### Disable Automatic Graceful Drain with AppStatus

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/07-module-structure.md

Example demonstrating how to disable automatic graceful drain handling using AppStatus. This is useful when manual control over shutdown flags is required.

```python
from sse_starlette.sse import AppStatus, ContentStream
from typing import AsyncGenerator

# Disable automatic shutdown handling
AppStatus.disable_automatic_graceful_drain()

async def my_stream() -> AsyncGenerator[dict, None]:
    # ...
```

--------------------------------

### Database Streaming SSE

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/08-quick-reference.md

Stream data directly from an asynchronous SQLAlchemy database session. This example demonstrates selecting active users and yielding their names as SSE events.

```python
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy import select

async def stream_users(request: Request):
    async with AsyncSession(engine) as session:
        stmt = select(User).where(User.active == True)
        result = await session.execute(stmt)
        
        for user in result.scalars():
            if await request.is_disconnected():
                break
            yield {"data": user.name, "id": str(user.id)}
            await asyncio.sleep(0.1)

@app.get("/users")
async def users(request: Request):
    return EventSourceResponse(stream_users(request))
```

--------------------------------

### Uvicorn Monkey-Patching for Graceful Shutdown

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/03-appstatus-shutdown.md

Monkey-patches uvicorn's Server.handle_exit to integrate the AppStatus graceful shutdown handler. This occurs at module import time and logs a debug message if uvicorn is not installed.

```python
try:
    from uvicorn.main import Server
    AppStatus.original_handler = Server.handle_exit
    Server.handle_exit = AppStatus.handle_exit
except ImportError:
    logger.debug("Uvicorn not installed. Graceful shutdown disabled.")
```

--------------------------------

### Timeout Protection Setup

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/05-configuration.md

Configure EventSourceResponse with a send_timeout to automatically close the connection if sending data stalls for a specified duration. Optionally, provide a callable to handle client disconnections.

```python
response = EventSourceResponse(
    stream(),
    send_timeout=30,  # Close connection if send stalls for 30s
    client_close_handler_callable=async_cleanup_on_disconnect
)
```

--------------------------------

### Configure Uvicorn for Production

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/08-quick-reference.md

Sets up Uvicorn with common production settings including host, port, workers, graceful shutdown timeout, and optimized loop/HTTP protocols.

```bash
uvicorn app:app \
  --host 0.0.0.0 \
  --port 8000 \
  --workers 4 \
  --timeout-graceful-shutdown 15 \
  --loop uvloop \
  --http httptools
```

--------------------------------

### Handle CancelledError in Generator

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/04-types-exceptions.md

Provides an example of how to catch and handle asyncio.CancelledError within a generator used by EventSourceResponse for cleanup.

```python
async def stream():
    try:
        while True:
            yield {"data": "tick"}
            await asyncio.sleep(1)
    except asyncio.CancelledError:
        # Cleanup code here
        print("Stream cancelled")
        raise  # Must re-raise
```

--------------------------------

### Run Tests

Source: https://github.com/sysid/sse-starlette/blob/main/README.md

Execute tests for the project. Use `make test-unit` for unit tests, `make test` for unit and integration tests, and `make test-experimentation` for optional experimentation tests.

```bash
make test-unit              # Unit tests only (default dev install)
```

```bash
make test                   # Unit + docker integration tests
```

```bash
make test-experimentation   # Optional experimentation tests (multi-consumer load tests)
                            # Requires the `experimentation` dependency group:
                            #   uv sync --group experimentation
```

--------------------------------

### Handle HTTP Disconnect Event

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/04-types-exceptions.md

Example of an async function to handle an http.disconnect message. It prints a message when the client disconnects.

```python
async def handle_disconnect(message: Message):
    # message = {"type": "http.disconnect"}
    print("Client disconnected")
```

--------------------------------

### Example Usage of SendTimeoutError

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/04-types-exceptions.md

Demonstrates setting a send_timeout on EventSourceResponse. If a send operation blocks for more than 5 seconds, SendTimeoutError will be raised.

```python
from sse_starlette import EventSourceResponse
from sse_starlette.sse import SendTimeoutError
import asyncio

@app.get("/events")
async def events():
    async def stream():
        for i in range(100):
            yield {"data": f"Event {i}"}
            await asyncio.sleep(0.5)
    
    return EventSourceResponse(
        stream(),
        send_timeout=5.0  # If send blocks > 5 seconds, raise SendTimeoutError
    )

# On client, if send times out:
# - The connection closes prematurely
# - No farewell event is sent
```

--------------------------------

### Minimal SSE Endpoint with Starlette

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/08-quick-reference.md

Demonstrates how to create a minimal SSE endpoint using Starlette directly. It sets up routes and uses EventSourceResponse to stream events.

```python
from starlette.applications import Starlette
from starlette.routing import Route
from sse_starlette import EventSourceResponse

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

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

--------------------------------

### Configure Multi-Line Data Separator

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/05-configuration.md

Demonstrates how the 'sep' parameter affects the output of multi-line data within ServerSentEvent objects.

```python
from sse_starlette import ServerSentEvent

event = ServerSentEvent(data="Line 1\nLine 2", sep="\r\n")
# Output: b'data: Line 1\r\ndata: Line 2\r\n\r\n'

event = ServerSentEvent(data="Line 1\nLine 2", sep="\n")
# Output: b'data: Line 1\ndata: Line 2\n\n'
```

--------------------------------

### EventSourceResponse enable_compression method

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/01-eventsourceresponse.md

Attempts to enable compression for SSE streams. Raises NotImplementedError as compression is not supported.

```python
def enable_compression(self, force: bool = False) -> None:
    ...
```

--------------------------------

### Type Checking with EventSourceResponse

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/07-module-structure.md

Demonstrates how to use type hints for static analysis when working with EventSourceResponse. Requires importing TYPE_CHECKING from typing.

```python
from typing import TYPE_CHECKING
from sse_starlette import EventSourceResponse

if TYPE_CHECKING:
    from sse_starlette.sse import ContentStream

def handler() -> EventSourceResponse:
    return EventSourceResponse(stream_data())
```

--------------------------------

### ServerSentEvent Constructor

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/02-serversentevent.md

Illustrates the parameters available for initializing a ServerSentEvent object, including data, event type, ID, retry interval, comments, and line separators.

```python
ServerSentEvent(
    data: Optional[Any] = None,
    *,
    event: Optional[str] = None,
    id: Optional[str] = None,
    retry: Optional[int] = None,
    comment: Optional[str] = None,
    sep: Optional[str] = None,
) -> None
```

--------------------------------

### JSONServerSentEvent with Complex Nested Data

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/02-serversentevent.md

An example of creating a `JSONServerSentEvent` with deeply nested JSON data. The entire structure is encoded as a JSON string in the `data` field of the SSE event.

```python
from sse_starlette import JSONServerSentEvent

event = JSONServerSentEvent(
    data={
        "alert": {
            "level": "warning",
            "message": "High memory usage",
            "metrics": {
                "memory_percent": 85.5,
                "memory_mb": 4096,
                "threshold_mb": 4800
            }
        }
    },
    event="system_alert",
    id="alert-2024-001"
)
encoded = event.encode()
# Client receives: {"alert":{"level":"warning",...}}
```

--------------------------------

### Advanced Import

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/08-quick-reference.md

Import EventSourceResponse and additional utilities for advanced SSE management, including app status, send timeouts, and content streams.

```python
from sse_starlette import EventSourceResponse
from sse_starlette.sse import AppStatus, SendTimeoutError, ContentStream
from sse_starlette.event import ensure_bytes
```

--------------------------------

### ServerSentEvent with Comments and Keepalives

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/02-serversentevent.md

Demonstrates using `ServerSentEvent` to send comments, which are ignored by clients but useful for server-side logging or keepalive pings. An example shows sending a timestamped comment.

```python
# Comment (invisible to client-side JavaScript)
ping_event = ServerSentEvent(
    comment="keepalive ping"
)
print(ping_event.encode())
# Output: b': keepalive ping\r\n\r\n'
```

```python
# Timestamp comment
import datetime
ping_event = ServerSentEvent(
    comment=f"ping: {datetime.datetime.utcnow().isoformat()}"
)
```

--------------------------------

### Custom Headers for Proxy Caching

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/01-eventsourceresponse.md

Configures EventSourceResponse with custom headers to control proxy caching behavior. This example sets `Cache-Control` to allow caching and adds a custom header.

```python
@app.get("/events")
async def cacheable_events():
    return EventSourceResponse(
        generate_events(),
        headers={
            "Cache-Control": "public, max-age=29",  # Allow proxy caching
            "X-Custom-Header": "value"
        }
    )
```

--------------------------------

### Run Uvicorn Server

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/INDEX.md

Command to run the FastAPI application using Uvicorn, a high-performance ASGI server. Includes graceful shutdown timeout configuration.

```bash
uvicorn app:app --timeout-graceful-shutdown 10
```

--------------------------------

### Error Handling: Catch in Generator

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/08-quick-reference.md

Implement resilient streaming by catching specific exceptions within the generator function. This example catches `ConnectionError` and yields an error event, then retries after a delay.

```python
async def resilient_stream(request):
    try:
        while True:
            try:
                data = await fetch_data()
                yield {"data": data}
            except ConnectionError as e:
                yield {"event": "error", "data": str(e)}
                await asyncio.sleep(5)
    except asyncio.CancelledError:
        raise
```

--------------------------------

### Simple Generator for SSE

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/06-architecture-patterns.md

Use this pattern for basic streaming where data is generated sequentially with delays between items. The generator yields data, and the connection closes cleanly when all items are exhausted.

```python
async def stream_data():
    for item in items:
        yield {"data": item, "id": str(item.id)}
        await asyncio.sleep(1)

@app.get("/events")
async def events():
    return EventSourceResponse(stream_data())
```

--------------------------------

### Import Internal Utility Module

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/04-types-exceptions.md

Imports the collapse_excgroups utility from the internal _utils module. This is not recommended for external use.

```python
from sse_starlette._utils import collapse_excgroups
```

--------------------------------

### Test SSE Endpoint with cURL

Source: https://github.com/sysid/sse-starlette/blob/main/examples/README.md

Use cURL to test the SSE endpoint. Replace '<endpoint>' with the actual endpoint defined in the example file. The '-N' flag disables output buffering.

```bash
curl -N http://localhost:8000/<endpoint>
```

--------------------------------

### Configure Uvicorn for Graceful Shutdown

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/08-quick-reference.md

Sets Uvicorn's graceful shutdown timeout and configures an SSE response with a matching grace period.

```bash
uvicorn app:app --timeout-graceful-shutdown 10
```

```python
response = EventSourceResponse(
    stream(),
    shutdown_event=shutdown_event,
    shutdown_grace_period=5.0  # Less than 10
)
```

--------------------------------

### Stream JSON Objects with EventSourceResponse

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/02-serversentevent.md

This example demonstrates streaming JSON objects as SSE events using `EventSourceResponse`. The `stream_users` async generator yields `JSONServerSentEvent` objects, and the endpoint returns an `EventSourceResponse`.

```python
import asyncio
from sse_starlette import JSONServerSentEvent, EventSourceResponse
from starlette.applications import Starlette
from starlette.routing import Route

app = Starlette(routes=[
    Route("/users", endpoint=lambda r: EventSourceResponse(stream_users()))
])

async def stream_users():
    users = [
        {"id": 1, "name": "Alice"},
        {"id": 2, "name": "Bob"},
        {"id": 3, "name": "Charlie"}
    ]
    
    for user in users:
        yield JSONServerSentEvent(
            data=user,
            event="user",
            id=str(user["id"])
        )
        await asyncio.sleep(1)
```

--------------------------------

### Multi-Threaded SSE Streaming

Source: https://github.com/sysid/sse-starlette/blob/main/README.md

Demonstrates running SSE streaming in separate threads using `asyncio.new_event_loop` and `asyncio.set_event_loop`. This avoids 'Event bound to different loop' errors.

```python
import threading
import asyncio
from sse_starlette import EventSourceResponse

def run_sse_in_thread():
    """SSE streaming works correctly in separate threads"""
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    
    async def thread_events():
        for i in range(5):
            yield {"data": f"Thread event {i}"}
            await asyncio.sleep(1)
    
    # This works without "Event bound to different loop" errors
    response = EventSourceResponse(thread_events())
    loop.close()

# Start SSE in multiple threads
for i in range(3):
    thread = threading.Thread(target=run_sse_in_thread)
    thread.start()
```

--------------------------------

### Thread-Safe Database Streaming

Source: https://github.com/sysid/sse-starlette/blob/main/README.md

Shows how to stream database results safely in a multi-threaded context by creating the session within the generator's context. Includes a check for client disconnection.

```python
async def stream_database_results(request):
    # CORRECT: Create session within generator context
    async with AsyncSession() as session:
        results = await session.execute(select(User))
        for row in results:
            if await request.is_disconnected():
                break
            yield {"data": row.name, "id": str(row.id)}

return EventSourceResponse(stream_database_results(request))
```

--------------------------------

### Configure Data Sender with AnyIO Memory Channels

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/05-configuration.md

Integrate with message queues using anyIO memory channels for push-based data sending. The data_sender_callable pushes data concurrently with the response stream.

```python
import anyio
from functools import partial

async def data_producer(send_channel):
    async with send_channel:
        for i in range(10):
            await send_channel.send({"data": f"Item {i}"})
            await anyio.sleep(1)

@app.get("/events")
async def stream_events():
    send_channel, receive_channel = anyio.create_memory_object_stream(10)
    
    return EventSourceResponse(
        receive_channel,
        data_sender_callable=partial(data_producer, send_channel)
    )
```

--------------------------------

### Production-Ready SSE Endpoint with FastAPI

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/INDEX.md

This snippet shows how to create a FastAPI endpoint that streams Server-Sent Events. It includes handling client disconnections and graceful shutdown.

```python
import anyio
from fastapi import FastAPI, Request
from sse_starlette import EventSourceResponse, ServerSentEvent
from datetime import datetime, timezone

app = FastAPI()

async def generate_events(request: Request):
    shutdown_event = anyio.Event()
    
    async def stream():
        counter = 0
        try:
            while not shutdown_event.is_set():
                if await request.is_disconnected():
                    break
                
                yield {
                    "data": f"Event #{counter}",
                    "id": str(counter),
                    "event": "update"
                }
                counter += 1
                
                # Wait for next event or shutdown signal
                with anyio.move_on_after(5.0):
                    await shutdown_event.wait()
            
            # Shutdown detected — send farewell
            yield ServerSentEvent(
                event="shutdown",
                data="Server closing"
            )
        except anyio.get_cancelled_exc_class():
            raise
    
    return EventSourceResponse(
        stream(),
        ping=30,
        send_timeout=60,
        shutdown_event=shutdown_event,
        shutdown_grace_period=5.0,
        headers={"Cache-Control": "no-cache"}
    )

@app.get("/events")
async def events(request: Request):
    return await generate_events(request)
```

--------------------------------

### Push-Based Streaming with Memory Channels

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/06-architecture-patterns.md

This pattern uses push-based streaming with a fixed-size memory channel buffer. The producer pushes data into the channel, and backpressure is handled automatically when the buffer is full. The producer runs as a background task.

```python
async def producer(send_channel):
    async with send_channel:
        for i in range(100):
            await send_channel.send({"data": f"Item {i}"})
            await asyncio.sleep(1)

@app.get("/events")
async def events():
    send_ch, recv_ch = anyio.create_memory_object_stream(10)
    
    return EventSourceResponse(
        recv_ch,
        data_sender_callable=partial(producer, send_ch)
    )
```

--------------------------------

### Broadcasting Pattern with SSE

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/08-quick-reference.md

Implement a broadcasting pattern to send messages to multiple connected clients using asyncio queues and EventSourceResponse.

```python
from fastapi import FastAPI
import asyncio

class Broadcaster:
    def __init__(self):
        self.clients = []
    
    def add_client(self):
        q = asyncio.Queue()
        self.clients.append(q)
        return q
    
    async def broadcast(self, message):
        for q in self.clients:
            await q.put({"data": message})

broadcaster = Broadcaster()

@app.get("/events")
async def subscribe():
    queue = broadcaster.add_client()
    
    async def stream():
        while True:
            yield await queue.get()
    
    return EventSourceResponse(stream())

@app.post("/send")
async def send_message(message: str):
    await broadcaster.broadcast(message)
```

--------------------------------

### Minimal SSE Endpoint with FastAPI

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/INDEX.md

This snippet demonstrates how to create a basic Server-Sent Events endpoint using FastAPI and sse-starlette. It requires importing `FastAPI` and `EventSourceResponse`.

```python
from fastapi import FastAPI
from sse_starlette import EventSourceResponse

app = FastAPI()

async def generate():
    for i in range(10):
        yield {"data": f"Event {i}"}

@app.get("/events")
async def stream():
    return EventSourceResponse(generate())
```

--------------------------------

### Optimize Network Settings for SSE

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/06-architecture-patterns.md

Adjust ping frequency and send timeout for EventSourceResponse to handle high connection counts and slow networks effectively. Consider reducing ping frequency for scenarios with many connections.

```python
# Reduce ping frequency for high-connection-count scenarios
response = EventSourceResponse(
    stream(),
    ping=45,  # Every 45 seconds instead of default 15
    send_timeout=120  # Longer timeout for slow networks
)
```

--------------------------------

### Testing SSE Stream Graceful Drain

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/03-appstatus-shutdown.md

Demonstrates how the automatic graceful drain of SSE streams functions in test environments using pytest. No manual setup is required as asyncio handles task cancellation during test teardown.

```python
import pytest
from sse_starlette import EventSourceResponse

@pytest.mark.asyncio
async def test_sse_stream():
    # No manual setup needed
    response = EventSourceResponse(generate_events())
    
    # During test teardown, asyncio handles cancellation
    # (no SIGTERM is sent, but task cancellation achieves similar effect)
```

--------------------------------

### Listen for Exit Signal

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/03-appstatus-shutdown.md

This function listens for an application exit signal. It registers an event with the shutdown state and waits for the event to be set, indicating that the application is shutting down. It ensures the watcher is started and unregisters the event upon completion.

```python
async def _listen_for_exit_signal() -> None:
    if AppStatus.should_exit:
        return  # Already shutting down
    
    _ensure_watcher_started_on_this_loop()  # Start watcher if needed
    
    state = _get_shutdown_state()
    event = anyio.Event()
    state.events.add(event)  # Register this stream
    
    try:
        if AppStatus.should_exit:
            return  # Double-check
        await event.wait()  # Block until shutdown
    finally:
        state.events.discard(event)  # Unregister
```

--------------------------------

### Manual Shutdown Handling in Custom Servers

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/03-appstatus-shutdown.md

For ASGI servers other than Uvicorn, you can manually enable automatic graceful drain and manage the shutdown process by setting `AppStatus.should_exit` and providing a grace period.

```python
# For non-uvicorn ASGI servers, you can still use manual shutdown
from sse_starlette.sse import AppStatus

# In your server startup
AppStatus.enable_automatic_graceful_drain()

# During graceful shutdown
async def shutdown():
    AppStatus.should_exit = True
    await asyncio.sleep(2)  # Grace period for streams

```

--------------------------------

### Basic Starlette SSE Endpoint

Source: https://github.com/sysid/sse-starlette/blob/main/README.md

A minimal Starlette application with an SSE endpoint that generates events every second. Requires `asyncio` and `starlette.applications.Starlette`.

```python
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)])
```

--------------------------------

### ServerSentEvent with Multi-Line Data

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/02-serversentevent.md

Illustrates how `ServerSentEvent` handles multi-line strings in the `data` field. Each line is prefixed with `data:` when encoded.

```python
event = ServerSentEvent(
    data="Line 1\nLine 2\nLine 3"
)
print(event.encode())
# Output: b'data: Line 1\r\ndata: Line 2\r\ndata: Line 3\r\n\r\n'
```

--------------------------------

### Basic Event Streaming with FastAPI

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/01-eventsourceresponse.md

Demonstrates how to stream Server-Sent Events using EventSourceResponse with a FastAPI application. The `generate_events` async generator yields data at intervals.

```python
from fastapi import FastAPI
from sse_starlette import EventSourceResponse
import asyncio

app = FastAPI()

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

@app.get("/events")
async def stream_events():
    return EventSourceResponse(generate_events())
```

--------------------------------

### Project Dependencies from pyproject.toml

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/07-module-structure.md

Specifies the minimum required Python version and core dependencies for the project.

```toml
[project]
requires-python = ">=3.10"

dependencies = [
    "starlette>=0.49.1",
    "anyio>=4.7.0",
]
```

--------------------------------

### Handle Slow Clients with Timeout

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/05-configuration.md

Illustrates a scenario where a tight send timeout can be triggered by a slow client receiving large data payloads.

```python
# This will timeout if client is slow to receive
async def stream():
    for i in range(1000):
        yield {"data": f"Large payload: " + "x" * 10000}  # Large data
        await asyncio.sleep(0.1)

response = EventSourceResponse(stream(), send_timeout=5)  # Tight timeout
```

--------------------------------

### Internal ExceptionGroup handling with collapse_excgroups

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/07-module-structure.md

The `collapse_excgroups` context manager is for internal use only and handles ExceptionGroup. It is a vendored functionality from Starlette and its stability is not guaranteed.

```python
# Don't do this — use Starlette's version if available
from sse_starlette._utils import collapse_excgroups
```

--------------------------------

### Send Simple String Event

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/08-quick-reference.md

Yield a simple string as an SSE message.

```python
yield "simple message"
```

--------------------------------

### ServerSentEvent Custom Event Creation

Source: https://github.com/sysid/sse-starlette/blob/main/README.md

Shows how to create a custom Server-Sent Event with specific data, event type, ID, and retry interval using the `ServerSentEvent` class.

```python
from sse_starlette import ServerSentEvent

event = ServerSentEvent(
    data="Custom message",
    event="notification", 
    id="msg-123",
    retry=5000
)
```

--------------------------------

### Configure Uvicorn Graceful Shutdown Timeout

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/INDEX.md

Set the '--timeout-graceful-shutdown' option for your ASGI server (e.g., Uvicorn) to be greater than the 'shutdown_grace_period' specified in EventSourceResponse for proper graceful shutdown.

```bash
uvicorn app:app --timeout-graceful-shutdown 10  # > 5.0
```

--------------------------------

### EventSourceResponse Properties and Methods

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/01-eventsourceresponse.md

This section details the configurable properties and core methods of the EventSourceResponse class, including how to set the ping interval and the ASGI application entrypoint.

```APIDOC
## EventSourceResponse

### Description

An ASGI-compatible response class for Server-Sent Events (SSE) that integrates with Starlette and FastAPI.

### Properties

#### `ping_interval`

- **Description:** The current ping interval in seconds.
- **Setter:** Validates and sets the ping interval. Must be a numeric value >= 0.

### Methods

#### `__call__`

##### Description

ASGI application entrypoint. This method is called by Starlette/FastAPI to handle the HTTP request. It orchestrates a task group that manages concurrent streaming, pings, shutdown detection, and client disconnect handling.

##### Parameters

- **`scope`** (`Scope`) - ASGI scope (contains request metadata)
- **`receive`** (`Receive`) - ASGI receive callable for incoming messages
- **`send`** (`Send`) - ASGI send callable for outgoing messages

##### Behavior

- Creates an `anyio` task group with five concurrent tasks: `_stream_response`, `_ping`, `_listen_for_exit_signal_with_grace`, `_listen_for_disconnect`, and an optional `data_sender_callable`.
- Wraps the task group in `collapse_excgroups()` for proper exception handling.
- Handles WebSocket denial by wrapping send messages.

##### Raises

Any exception from the generator or send operations (except `asyncio.CancelledError` which is handled gracefully).

#### `enable_compression`

##### Description

Attempts to enable compression for SSE streams.

##### Raises

- `NotImplementedError` - Compression is not supported for SSE streams.
```

--------------------------------

### Synchronous Generator for SSE

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/04-types-exceptions.md

Demonstrates a synchronous generator that yields content suitable for Server-Sent Events.

```python
# Sync generator
def stream_sync() -> Iterator[Content]:
    yield "item 1"
    yield "item 2"
```

--------------------------------

### SSE Task Group Architecture with `cancel_on_finish`

Source: https://github.com/sysid/sse-starlette/blob/main/ARCHITECTURE.md

Illustrates the concurrent task execution within `EventSourceResponse.__call__`. It uses `anyio.create_task_group` with `cancel_on_finish` to ensure that whichever task completes first cancels all other sibling tasks. This pattern is crucial for managing the lifecycle of SSE connections.

```python
EventSourceResponse.__call__(scope, receive, send)
   |
   v
anyio.create_task_group()
   |
   +-- cancel_on_finish(_stream_response)      # pushes SSE data to client
   +-- cancel_on_finish(_ping)                 # keepalive pings every ~15s
   +-- cancel_on_finish(_listen_for_exit_signal_with_grace)  # server shutdown
   +-- cancel_on_finish(_listen_for_disconnect) # client disconnect
   |   (+ optional: data_sender_callable)
   |
   v
All tasks cancelled --> background task (if any) --> return
```

--------------------------------

### Minimal SSE Starlette Import

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/INDEX.md

Import the essential EventSourceResponse class for basic SSE functionality.

```python
from sse_starlette import EventSourceResponse
```

--------------------------------

### Create EventSourceResponse

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/INDEX.md

Instantiate EventSourceResponse with content, ping interval, and optional timeouts or shutdown events. Use this to handle SSE streams in ASGI applications.

```python
response = EventSourceResponse(
    content,                    # Async iterable or sync iterator
    ping=15,                    # Seconds between pings (0 to disable)
    send_timeout=None,          # Timeout for send operations
    shutdown_event=None,        # anyio.Event for graceful shutdown
    shutdown_grace_period=0,    # Seconds after shutdown signal
    # ... plus headers, ping_message_factory, etc.
)
```

--------------------------------

### SSE with Graceful Shutdown Handling

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/07-module-structure.md

Implement graceful shutdown by providing a shutdown event and grace period. This ensures ongoing events are completed or handled before the connection closes.

```python
import anyio
from sse_starlette import EventSourceResponse

async def graceful():
    shutdown_event = anyio.Event()
    
    async def stream():
        try:
            while not shutdown_event.is_set():
                yield {"data": "tick"}
                with anyio.move_on_after(1.0):
                    await shutdown_event.wait()
            yield {"event": "shutdown"}
        except anyio.get_cancelled_exc_class():
            raise
    
    return EventSourceResponse(
        stream(),
        shutdown_event=shutdown_event,
        shutdown_grace_period=5.0
    )

@app.get("/events")
async def events():
    return await graceful()
```

--------------------------------

### Format arbitrary data as SSE with ServerSentEvent

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/07-module-structure.md

Use `ServerSentEvent` to format arbitrary data into the Server-Sent Events format. This is a standard way to create SSE messages.

```python
from sse_starlette import ServerSentEvent, JSONServerSentEvent

event = ServerSentEvent(data="Hello", event="greeting")
json_event = JSONServerSentEvent(data={"key": "value"})
```

--------------------------------

### Graceful Shutdown with Cooperative Closure

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/01-eventsourceresponse.md

Implements graceful shutdown for SSE streams by using an `anyio.Event` to signal closure. The stream cooperates by sending a final message before terminating.

```python
import anyio
from sse_starlette import EventSourceResponse

async def graceful_stream():
    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 final message
            yield {"event": "shutdown", "data": "Server closing"}
        except anyio.get_cancelled_exc_class():
            raise
    
    return EventSourceResponse(
        generate(),
        shutdown_event=shutdown_event,
        shutdown_grace_period=5.0
    )
```

--------------------------------

### EventSourceResponse Basic Usage

Source: https://github.com/sysid/sse-starlette/blob/main/README.md

Demonstrates the basic usage of `EventSourceResponse` by yielding dictionaries containing 'data', 'event', and 'id' fields. Ensure `data` is an iterable.

```python
from sse_starlette import EventSourceResponse

# Basic usage
async def stream_data():
    for item in data:
        yield {"data": item, "event": "update", "id": str(item.id)}

return EventSourceResponse(stream_data())
```