WebSocket、流式响应与进阶方向

先按通信需求选择工具

需求	可以考虑
一次查询、一次提交	普通 HTTP 接口
逐步返回大内容或生成结果	流式 HTTP 响应
服务端持续向客户端推送事件	SSE
客户端和服务端持续双向交互	WebSocket

不要为了显得高级就给所有接口使用长连接。多数 CRUD 使用普通 HTTP 更容易维护。

完整示例：WebSocket 回声

保存为 realtime_demo.py：

from fastapi import FastAPI, WebSocket, WebSocketDisconnect

app = FastAPI()


@app.websocket("/ws")
async def websocket_echo(websocket: WebSocket):
    await websocket.accept()
    try:
        while True:
            text = await websocket.receive_text()
            await websocket.send_text(f"服务器收到：{text}")
    except WebSocketDisconnect:
        pass

启动：

python -m uvicorn realtime_demo:app --reload

accept() 接受连接；receive_text() 等待文本；send_text() 发送文本。地址是 ws://127.0.0.1:8000/ws，HTTPS 环境通常使用 wss://。官方 WebSocket 说明

普通浏览器地址栏和 /docs 不提供与 HTTP 接口相同的 WebSocket 调用方式。可以使用专用客户端，或者用下面的测试直接验证：

from fastapi.testclient import TestClient
from realtime_demo import app


with TestClient(app) as client:
    with client.websocket_connect("/ws") as connection:
        connection.send_text("你好")
        assert connection.receive_text() == "服务器收到：你好"

真实聊天室还需要用户认证、断线清理、消息格式、心跳、容量限制。多个工作进程之间广播消息时，仅使用 Python 列表保存连接并不够，需要进程间通信方案。

追加示例：流式文本

追加到 realtime_demo.py：

import asyncio
from fastapi.responses import StreamingResponse


async def generate_lines():
    for number in range(1, 4):
        yield f"第 {number} 条内容\n"
        await asyncio.sleep(0.2)


@app.get("/stream")
async def stream_lines():
    return StreamingResponse(generate_lines(), media_type="text/plain")

生成器逐步产生内容，响应可以逐步发送。客户端显示、反向代理缓冲和网络分块会影响你看到的效果；不要把每一次 yield 都当成客户端一定立即收到的独立消息。官方流式响应说明

适合后续学习的主题

方向	核心知识	什么时候学
SQL 基础	关系、连接查询、事务、索引	开始数据库实战时
数据迁移	Alembic、变更版本、回滚策略	表结构开始变化时
异步数据库	异步驱动、Tortoise 连接池、事务范围	有明确异步需求时
缓存	失效策略、Redis、一致性	查询性能出现需求时
任务队列	持久任务、重试、幂等	工作必须可靠完成时
权限	资源归属、角色、权限范围	项目有多用户时
可观测性	结构化日志、指标、链路追踪	开始部署运行时
API 版本	兼容性、弃用流程、客户端迁移	接口有稳定调用方时
性能分析	测量瓶颈、查询计划、连接复用	有实际性能问题时

本章练习

运行回声测试；请求 /stream 并观察响应。分别给 HTTP、SSE 和 WebSocket 想一个适合的使用场景。

过关标准：能根据真实通信需求选择方案，而不是只根据技术名字选择。