FastAPI SSE WebSocket 实战教程第一章 项目初始化与环境搭建1.1 使用 uv 初始化项目uv是新一代 Python 项目管理和包管理工具速度比 pip 快 10-100 倍。# 初始化项目--app 表示这是一个应用项目--name 指定包名uv init--app--namedemo-app执行后生成pyproject.toml这是项目的配置文件记录项目元数据和依赖信息。1.2 安装依赖使用uv add安装依赖不指定版本号会自动安装最新版本uvaddfastapi uvicorn websockets各依赖的作用fastapiWeb 框架提供路由、请求处理等核心功能uvicornASGI 服务器负责运行 FastAPI 应用websocketsWebSocket 协议支持库安装完成后项目根目录会生成.venv虚拟环境目录和uv.lock依赖锁定文件。1.3 项目目录结构项目根目录/ ├── src/ │ └── main.py # Python 主文件后端代码 ├── templates/ │ └── index.html # HTML 页面前端代码 ├── pyproject.toml # 项目配置文件 ├── uv.lock # 依赖锁定文件 └── .venv/ # 虚拟环境第二章 创建 FastAPI 应用2.1 导入模块并创建实例importasyncioimportdatetimefrompathlibimportPathimportuvicornfromfastapiimportFastAPI,WebSocket,WebSocketDisconnectfromfastapi.responsesimportStreamingResponse,HTMLResponse appFastAPI()FastAPI()创建应用实例所有路由和功能都注册在这个实例上WebSocketWebSocket 连接的类提供收发消息的方法WebSocketDisconnect客户端断开连接时触发的异常需要捕获处理StreamingResponse流式响应用于 SSE 场景HTMLResponse返回 HTML 内容的响应类型2.2 启动服务if__name____main__:configuvicorn.Config(app,host127.0.0.1,port8000)serveruvicorn.Server(config)asyncio.run(server.serve())uvicorn.Configuvicorn.Server是比uvicorn.run()更现代的启动方式Config封装所有配置参数主机、端口、日志级别等Server提供精细化的生命周期控制优雅关闭等asyncio.run()运行异步入口运行命令.venv\Scripts\python src\main.py打开浏览器访问http://127.0.0.1:8000会看到空白页面因为我们还没写路由终端会输出INFO: Started server process [xxxxx] INFO: Waiting for application startup. INFO: Application startup complete. INFO: Uvicorn running on http://127.0.0.1:8000第三章 返回静态页面3.1 为什么需要独立 HTML 文件将 HTML 代码从 Python 中分离出来好处很明显前后端解耦修改样式无需改动 Python 代码HTML 文件获得 IDE 的高亮、补全支持文件结构清晰维护方便3.2 编写首页路由app.get(/)asyncdefindex():html_pathPath(__file__).parent.parent/templates/index.htmlhtml_contenthtml_path.read_text(encodingutf-8)returnHTMLResponse(html_content)app.get(/)装饰器将函数注册为 HTTP GET 请求的处理函数Path(__file__)当前 Python 文件src/main.py的路径.parent.parent上两级目录即项目根目录read_text(encodingutf-8)以 UTF-8 编码读取文件内容HTMLResponse()将字符串包装为 HTML 响应返回给浏览器这样访问http://127.0.0.1:8000时服务端读取templates/index.html文件内容并返回给浏览器渲染。第四章 异步编程基础在进入 SSE 和 WebSocket 之前需要先理解异步编程的几个核心概念。4.1 async / awaitimportasyncioasyncdefmy_function():# async 定义异步函数print(开始执行)awaitasyncio.sleep(1)# await 等待异步操作完成print(1秒后执行)async def定义异步函数协程调用时返回一个协程对象不会立即执行await等待一个异步操作完成同时让出事件循环让其他任务执行4.2 异步生成器asyncdefmy_generator():foriinrange(5):awaitasyncio.sleep(1)yieldi# yield 每次返回一个值函数暂停下次继续async defyield定义异步生成器每次yield返回一个值后函数暂停执行下次迭代时从yield之后继续这使得函数可以边生产边消费不需要一次性生成所有数据第五章 SSE 服务端推送5.1 SSE 协议简介SSEServer-Sent Events服务端推送事件是一种轻量级的实时通信协议基于 HTTP 协议使用简单服务端可以主动向客户端推送数据客户端使用EventSourceAPI 接收适合单向推送场景通知、行情、日志等SSE 数据格式data: 消息内容\n\n必须以data:开头\n\n两个换行结尾。5.2 编写 SSE 事件生成器asyncdefevent_generator():whileTrue:nowdatetime.datetime.now().strftime(%H:%M:%S)yieldfdata: 当前时间:{now}\n\nawaitasyncio.sleep(1)while True持续循环每个连接生命周期内运行datetime.datetime.now().strftime(%H:%M:%S)获取当前时间并格式化为HH:MM:SSyield每次产生一个 SSE 事件后暂停等待下一次迭代await asyncio.sleep(1)休眠 1 秒让出事件循环给其他请求处理关键理解这不是死循环。每次yield后函数暂停等待StreamingResponse消费sleep期间事件循环可处理其他请求客户端断开时生成器被自动取消。5.3 注册 SSE 路由app.get(/sse)asyncdefsse():returnStreamingResponse(event_generator(),media_typetext/event-stream)app.get(/sse)注册 GET 请求路由StreamingResponse()将异步生成器包装为流式 HTTP 响应media_typetext/event-stream设置 Content-Type浏览器识别此类型后会保持连接并持续接收5.4 前端接收 SSEconstevtSourcenewEventSource(/sse);evtSource.addEventListener(message,(e){sseDiv.innerHTMLdiv${e.data}/div;});EventSource(/sse)创建 SSE 连接自动连接到/sse端点addEventListener(message, ...)监听message事件每次服务端推送触发e.data服务端推送的数据内容innerHTML 覆盖显示最新一条数据替换之前的第六章 WebSocket 双向通信6.1 WebSocket 协议简介WebSocket 与 SSE 的核心区别特性SSEWebSocket方向仅服务端→客户端双向通信协议HTTP独立协议ws://数据格式文本文本/二进制自动重连内置支持需手动实现适用场景通知、推送聊天、游戏、协作6.2 编写 WebSocket 端点app.websocket(/ws)asyncdefwebsocket_endpoint(websocket:WebSocket):awaitwebsocket.accept()try:asyncfordatainwebsocket.iter_text():awaitwebsocket.send_text(f收到:{data})exceptWebSocketDisconnect:passapp.websocket(/ws)注册 WebSocket 路由await websocket.accept()接受客户端的握手请求完成 HTTP → WebSocket 协议升级async for data in websocket.iter_text()iter_text()是 Starlette 提供的异步迭代器逐个接收客户端发来的文本消息客户端断开时会自动抛出WebSocketDisconnect比while True: data await websocket.receive_text()更简洁await websocket.send_text(f收到: {data})向客户端发送文本消息except WebSocketDisconnect捕获客户端断开异常避免服务端报错6.3 前端 WebSocket 客户端constwsnewWebSocket(ws://127.0.0.1:8000/ws);ws.addEventListener(open,(){wsDiv.innerHTMLdiv stylecolor:green已连接/div;});ws.addEventListener(message,(e){wsDiv.innerHTMLdiv${e.data}/div;});functionsend(){constinputdocument.getElementById(msg);ws.send(input.value);input.value;}WebSocket(ws://...)创建 WebSocket 连接addEventListener(open, ...)连接建立成功时触发addEventListener(message, ...)收到服务端消息时触发ws.send(input.value)向服务端发送消息使用addEventListener而非onmessage/onopen属性赋值是更现代的 JavaScript 实践支持绑定多个监听器第七章 启动方式演进7.1 传统方式uvicorn.run()# 旧写法仍然可用if__name____main__:uvicorn.run(app,host127.0.0.1,port8000)uvicorn.run()是一个便捷函数内部封装了Config和Server的创建和启动逻辑。7.2 现代方式uvicorn.Config Server# 新写法推荐if__name____main__:configuvicorn.Config(app,host127.0.0.1,port8000)serveruvicorn.Server(config)asyncio.run(server.serve())优势更灵活可以访问Config和Server对象进行精细控制优雅关闭Server支持should_exit信号可以平滑关闭便于扩展可以同时运行多个服务或与其他异步任务一起启动第八章 完整项目代码8.1 src/main.pyFastAPI SSE WebSocket 最简单案例最新语法importasyncioimportdatetimefrompathlibimportPathimportuvicornfromfastapiimportFastAPI,WebSocket,WebSocketDisconnectfromfastapi.responsesimportStreamingResponse,HTMLResponse# 创建 FastAPI 应用实例appFastAPI()# SSE 服务端推送 asyncdefevent_generator(): 异步生成器每秒产生一个 SSE 事件 SSE 格式data: 内容\n\n两个换行结尾 whileTrue:# 获取当前时间并格式化为 HH:MM:SSnowdatetime.datetime.now().strftime(%H:%M:%S)# yield 返回 SSE 数据行data: 前缀 两个换行是 SSE 协议要求yieldfdata: 当前时间:{now}\n\n# 休眠 1 秒避免无限循环占用 CPUawaitasyncio.sleep(1)app.get(/sse)asyncdefsse(): SSE 端点 StreamingResponse 将异步生成器的内容以 text/event-stream 格式流式返回 浏览器端的 EventSource 会自动解析此格式 returnStreamingResponse(event_generator(),media_typetext/event-stream)# WebSocket 双向通信 app.websocket(/ws)asyncdefwebsocket_endpoint(websocket:WebSocket): WebSocket 端点 app.websocket 装饰器将函数注册为 WebSocket 处理器 # 接受客户端的 WebSocket 握手请求完成 HTTP 升级awaitwebsocket.accept()try:# iter_text() 是 Starlette 提供的异步迭代器# 它会逐个接收客户端发来的文本消息# 当客户端断开时自动抛 WebSocketDisconnect无需手动 while Trueasyncfordatainwebsocket.iter_text():# 将收到的消息原样返回回声# 使用 f-string 构造回复字符串awaitwebsocket.send_text(f收到:{data})exceptWebSocketDisconnect:# 客户端断开连接时捕获此异常非错误正常行为pass# 首页 app.get(/)asyncdefindex(): 首页返回 HTML 测试页面 Path(__file__).parent.parent 定位到项目根目录src/ 的上级 # 构建 HTML 文件的绝对路径html_pathPath(__file__).parent.parent/templates/index.html# 读取 HTML 文件内容UTF-8 编码html_contenthtml_path.read_text(encodingutf-8)# 包装为 HTMLResponse 返回returnHTMLResponse(html_content)# 入口 if__name____main__:# 使用 uvicorn.Config uvicorn.Server 方式启动# 相比 uvicorn.run()这种方式支持更灵活的控制和优雅关闭configuvicorn.Config(app,host127.0.0.1,port8000)serveruvicorn.Server(config)# asyncio.run() 运行异步入口asyncio.run(server.serve())8.2 templates/index.html!DOCTYPEhtmlhtmllangzh-CNheadmetacharsetUTF-8titleSSE WebSocket 测试/title/headbody!-- SSE 测试区域 --h2SSE 测试/h2!-- SSE 消息会动态追加到这个 div 中 --dividssestyleborder:1px solid #ccc;padding:10px;min-height:60px;/div!-- WebSocket 测试区域 --h2WebSocket 测试/h2!-- 用户输入消息的文本框 --inputidmsgplaceholder输入消息/!-- 点击按钮发送消息 --buttononclicksend()发送/button!-- WebSocket 消息会动态追加到这个 div 中 --dividwsstyleborder:1px solid #ccc;padding:10px;margin-top:10px;min-height:60px;/divscript// SSE 客户端 // 获取 SSE 显示区域的 DOM 元素constsseDivdocument.getElementById(sse);// 创建 EventSource 对象连接到 /sse 端点constevtSourcenewEventSource(/sse);// 监听 message 事件每当服务端推送 SSE 事件时触发evtSource.addEventListener(message,(e){// e.data 包含服务端发送的数据即 当前时间: HH:MM:SS// 将数据覆盖显示替换上一次结果sseDiv.innerHTMLdiv${e.data}/div;});// WebSocket 客户端 // 获取 WebSocket 显示区域的 DOM 元素constwsDivdocument.getElementById(ws);// 创建 WebSocket 连接连接到 ws://127.0.0.1:8000/wsconstwsnewWebSocket(ws://127.0.0.1:8000/ws);// 监听 message 事件服务端通过 WebSocket 发送消息时触发ws.addEventListener(message,(e){// e.data 包含服务端返回的文本wsDiv.innerHTMLdiv${e.data}/div;});// 监听 open 事件WebSocket 连接建立成功时触发ws.addEventListener(open,(){// 在显示区域追加绿色提示文字wsDiv.innerHTMLdiv stylecolor:green已连接/div;});// 发送消息函数点击按钮时调用functionsend(){// 获取输入框中的文本constinputdocument.getElementById(msg);// 通过 WebSocket 发送文本到服务端ws.send(input.value);// 清空输入框方便下一条消息input.value;}/script/body/html8.3 pyproject.toml[project] name demo-app version 0.1.0 description FastAPI SSE WebSocket 演示项目 readme README.md requires-python 3.14 dependencies [ fastapi, uvicorn, websockets, ]第九章 运行与测试9.1 启动服务cd项目根目录 .venv\Scripts\python src\main.py终端输出INFO: Started server process [xxxxx] INFO: Waiting for application startup. INFO: Application startup complete. INFO: Uvicorn running on http://127.0.0.1:80009.2 测试 SSE打开浏览器访问http://127.0.0.1:8000页面顶部 SSE 区域会每秒自动更新当前时间打开浏览器开发者工具 → 网络标签可以看到一个持续连接9.3 测试 WebSocket在页面 WebSocket 区域的输入框中输入文字点击发送按钮服务端会返回收到: 你输入的文字消息记录会累积显示在 div 中第十章 知识总结核心知识点知识点说明uv init使用 uv 初始化 Python 项目uv add安装依赖不指定版本自动安装最新版app.get()HTTP GET 路由装饰器app.websocket()WebSocket 路由装饰器StreamingResponse流式响应用于 SSEHTMLResponseHTML 内容响应async/await异步编程语法async for异步迭代器用于 WebSocket 消息接收WebSocketDisconnectWebSocket 断开异常uvicorn.ConfigServer现代启动方式EventSource浏览器端 SSE 客户端 APIWebSocket浏览器端 WebSocket 客户端 APIaddEventListener现代 JavaScript 事件监听方式学习路径建议先掌握 HTTP 基础理解 GET/POST 请求-响应模型再学习异步编程理解 async/await 和事件循环然后学习 SSE单向推送概念简单容易上手最后学习 WebSocket双向通信概念稍复杂进阶方向WebSocket 广播、房间管理、身份认证、心跳检测