如何判断FastAPI/Starlette中StreamingResponse的连接是否已断开？

针对EventSource场景下需要感知客户端断开连接的需求，以下是几种实用的实现方式：

1. 捕获生成器迭代时的异常 #

当客户端主动断开连接，Starlette在尝试向客户端发送数据时会抛出连接相关异常（如ConnectionResetError、BrokenPipeError），同时生成器会收到GeneratorExit信号。只需在生成器内部捕获这些异常，即可触发断开事件处理逻辑。

示例代码：

from fastapi import FastAPI
from starlette.responses import StreamingResponse
import asyncio

app = FastAPI()

async def event_stream():
    try:
        while True:
            # 模拟发送EventStream格式的数据
            yield f"data: {asyncio.get_event_loop().time()}\n\n"
            await asyncio.sleep(1)
    except (ConnectionResetError, BrokenPipeError, GeneratorExit):
        # 客户端断开时执行的逻辑
        print("客户端已断开连接")
        # 替换为你的自定义处理，比如清理资源、通知其他服务等
        cleanup_client_connection()

def cleanup_client_connection():
    # 自定义断开处理逻辑示例
    pass

@app.get("/stream")
async def stream():
    return StreamingResponse(event_stream(), media_type="text/event-stream")

2. 主动检查Request连接状态 #

通过FastAPI的Request对象，可以调用异步方法is_disconnected()主动检测客户端连接状态。适合发送数据间隔较长的场景，无需等待异常触发就能及时感知断开。

示例代码：

from fastapi import FastAPI, Request
from starlette.responses import StreamingResponse
import asyncio

app = FastAPI()

async def event_stream(request: Request):
    while True:
        # 主动检查连接是否已断开
        if await request.is_disconnected():
            print("客户端已断开连接")
            cleanup_client_connection()
            break
        yield f"data: {asyncio.get_event_loop().time()}\n\n"
        await asyncio.sleep(1)

def cleanup_client_connection():
    pass

@app.get("/stream")
async def stream(request: Request):
    return StreamingResponse(event_stream(request), media_type="text/event-stream")

3. 异步上下文管理器实现优雅清理 #

使用异步上下文管理器可以将断开后的资源清理逻辑封装起来，无论连接是正常结束还是异常断开，都会自动执行清理操作，代码结构更整洁。

示例代码：

from fastapi import FastAPI, Request
from starlette.responses import StreamingResponse
import asyncio
from contextlib import asynccontextmanager

app = FastAPI()

@asynccontextmanager
async def connection_handler():
    try:
        yield
    finally:
        # 连接终止时统一执行的清理逻辑
        print("客户端连接已终止")
        cleanup_client_connection()

async def event_stream(request: Request):
    async with connection_handler():
        while True:
            if await request.is_disconnected():
                break
            yield f"data: {asyncio.get_event_loop().time()}\n\n"
            await asyncio.sleep(1)

def cleanup_client_connection():
    pass

@app.get("/stream")
async def stream(request: Request):
    return StreamingResponse(event_stream(request), media_type="text/event-stream")

内容的提问来源于stack exchange，提问作者est