FastAPI 生产接口：依赖、安全、生命周期与流式响应

本篇目标

看懂真实 FastAPI 项目中路由之外的关键结构：依赖注入怎样组装请求资源、认证和授权怎样分层、全局资源怎样启动关闭，以及流式输出怎样控制生命周期。

使用依赖注入区分认证与授权管理应用生命周期实现流式响应

先记住一句话

生产 FastAPI 路由只负责 HTTP 边界；身份、Session 和服务通过依赖注入获得，连接池和客户端由 lifespan 统一管理，流式请求则必须处理断连、超时与下游取消。

一次请求经过哪些层

HTTP 请求
└─ ASGI Server（例如 Uvicorn）
   └─ Middleware：请求 ID、日志、CORS 等横切逻辑
      └─ FastAPI 路由匹配与 Pydantic 校验
         └─ Depends 依赖树
            ├─ 认证当前调用方
            ├─ 创建请求级 Session
            └─ 组装 Service
               └─ 执行业务用例并返回领域结果
                  └─ response_model 过滤并序列化响应

Middleware 适合对几乎所有请求都生效、且不依赖具体业务参数的逻辑。依赖适合认证、权限和请求级资源。业务规则仍应放在 Service，而不是塞进 Middleware 或路由装饰器。

依赖注入不是全局变量

# 文件位置：src/note_api/dependencies.py
from typing import Annotated

from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession

from .database import get_session
from .service import NoteService

def get_note_service(
    # Annotated 保留 AsyncSession 类型，并附加 Depends 元数据供 FastAPI 注入。
    session: Annotated[AsyncSession, Depends(get_session)],
) -> NoteService:
    # FastAPI 先解析 get_session，再把结果传入当前工厂。
    return NoteService(session)

# 类型别名减少路由中的重复声明，同时保留编辑器类型信息。
NoteServiceDep = Annotated[NoteService, Depends(get_note_service)]

Depends 描述的是 “当前函数需要什么” ，FastAPI 负责解析依赖树和作用域。测试时可以覆盖依赖工厂，从而替换数据库或外部客户端，而不需要修改业务代码。

认证与授权要分开

• 认证（Authentication）：确认调用方是谁；
• 授权（Authorization）：确认这个调用方能否对当前资源执行该动作；
• 资源过滤：查询时就加入 owner_id 或租户条件，不能先取出任意数据再靠前端隐藏。

下面是内部服务使用 API Key 的最小示例。面向用户的登录系统通常应接入成熟的 OAuth2 / OpenID Connect 身份提供方，不能把这个示例当成完整账户系统。

# 文件位置：src/note_api/auth.py
import hmac
import os
from dataclasses import dataclass
from typing import Annotated

from fastapi import Depends, HTTPException, status
from fastapi.security import APIKeyHeader

# 从 X-API-Key 请求头读取密钥；auto_error=False 让函数统一决定错误响应。
api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False)

# frozen=True 表示认证结果创建后不能通过普通赋值修改。
@dataclass(frozen=True)
class Principal:
    subject: str  # 经过认证的调用方标识，不等于任意请求参数。

def require_principal(
    provided_key: Annotated[str | None, Depends(api_key_header)],
) -> Principal:
    # 从进程环境变量读取服务端密钥，避免把秘密硬编码进仓库。
    expected_key = os.environ.get("NOTE_API_KEY")
    if expected_key is None:
        # 服务端没有配置密钥属于部署错误，不能降级为匿名访问。
        raise RuntimeError("NOTE_API_KEY is not configured")

    if provided_key is None or not hmac.compare_digest(provided_key, expected_key):
        # compare_digest 避免普通字符串比较暴露明显的时间差异。
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="invalid API key",
        )

    return Principal(subject="internal-client")

PrincipalDep = Annotated[Principal, Depends(require_principal)]

API Key 只能说明持有者拥有一段共享秘密，通常缺少用户级撤销、细粒度权限和标准登录流程。JWT 是签名令牌而不是加密容器，不能放入密码等敏感明文；服务还必须验证签名算法、签发方、受众、过期时间和权限声明。

路由只映射 HTTP 与业务

# 文件位置：src/note_api/routes/notes.py
from fastapi import APIRouter, HTTPException, status
from pydantic import BaseModel, Field

from ..auth import PrincipalDep
from ..dependencies import NoteServiceDep
from ..service import CreateNoteCommand

# Router 把同一组接口集中管理；prefix 会自动加到本文件的所有路由前。
router = APIRouter(prefix="/notes", tags=["notes"])

# BaseModel 会按字段标注校验和序列化外部数据；Field 声明长度约束。
class CreateNoteRequest(BaseModel):
    title: str = Field(min_length=1, max_length=200)
    content: str = Field(min_length=1, max_length=50_000)

class NoteResponse(BaseModel):
    id: str
    title: str
    content: str

# 路由装饰器把下面的函数注册为 POST /notes，并限制响应结构与成功状态码。
@router.post("", response_model=NoteResponse, status_code=status.HTTP_201_CREATED)
async def create_note(
    body: CreateNoteRequest,
    principal: PrincipalDep,
    service: NoteServiceDep,
) -> NoteResponse:
    note = await service.create(
        CreateNoteCommand(
            owner_id=principal.subject,  # 身份来自认证结果，不信任 body 中的用户 ID。
            title=body.title,
            content=body.content,
        )
    )
    return NoteResponse(id=note.id, title=note.title, content=note.content)

# {note_id} 是路径参数，FastAPI 会把 URL 中的对应部分传给同名形参。
@router.get("/{note_id}", response_model=NoteResponse)
async def get_note(
    note_id: str,
    principal: PrincipalDep,
    service: NoteServiceDep,
) -> NoteResponse:
    note = await service.get_for_owner(note_id, principal.subject)
    if note is None:
        # 对无权限和不存在都返回 404，可避免泄露其他用户资源是否存在。
        raise HTTPException(status_code=404, detail="note not found")
    return NoteResponse(id=note.id, title=note.title, content=note.content)

lifespan 管理全应用资源

数据库 Engine、HTTP 连接池和大型模型应在应用启动时创建一次，并在应用关闭时释放。FastAPI 当前推荐使用 lifespan 上下文管理器表达这段生命周期。

# 文件位置：src/note_api/main.py
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager

import httpx
from fastapi import FastAPI

from .database import engine
from .routes.notes import router as notes_router

# @asynccontextmanager 把含 yield 的异步生成器转换为应用生命周期管理器。
@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    # AsyncClient 复用连接池；超时必须显式设置，不能无限等待下游。
    app.state.http_client = httpx.AsyncClient(timeout=10.0)
    try:
        yield  # yield 之前是启动阶段，之后应用开始接收请求。
    finally:
        await app.state.http_client.aclose()  # 先停止外部 HTTP 客户端。
        await engine.dispose()  # 再释放数据库连接池。

app = FastAPI(title="Note API", lifespan=lifespan)
app.include_router(notes_router)

不要同时混用 lifespan 和旧的 startup / shutdown 事件来管理同一批资源。多 Worker 部署时，每个进程都有独立的内存和连接池，不能假设 app.state 在进程间共享。

流式响应不是普通 JSON 分多次返回

StreamingResponse 接收同步或异步迭代器，逐步写出响应体。聊天接口常使用 SSE 格式传输事件，但业务仍要处理客户端断开、生成取消、错误事件和最终完成状态。

# 文件位置：src/note_api/routes/stream.py
import asyncio
import json
from collections.abc import AsyncIterator

from fastapi import APIRouter, Request
from fastapi.responses import StreamingResponse

router = APIRouter(prefix="/stream", tags=["stream"])

async def generate_events(request: Request) -> AsyncIterator[str]:
    # AsyncIterator[str] 表示函数会通过 yield 异步、逐段产生字符串事件。
    for index in range(3):
        if await request.is_disconnected():
            return  # 客户端已离开，不再继续消耗下游资源。

        await asyncio.sleep(0.2)  # 模拟异步模型或检索调用。
        payload = json.dumps({"index": index, "text": f"chunk-{index}"})
        yield f"event: message\ndata: {payload}\n\n"  # SSE 事件以空行结束。

    yield "event: done\ndata: {}\n\n"  # 显式告诉客户端本次生成已完成。

# 返回 StreamingResponse 后，服务器会边生成边发送，而不是等全部内容完成。
@router.get("/notes")
async def stream_notes(request: Request) -> StreamingResponse:
    return StreamingResponse(
        generate_events(request),
        media_type="text/event-stream",
        headers={"Cache-Control": "no-cache"},
    )

流式响应开始发送后，已经不能把中途错误改成新的 HTTP 状态码，通常要在事件协议中定义 error 和 done。如果反向代理启用了缓冲，还可能导致事件攒到最后才到浏览器。

后台任务和任务队列的边界

FastAPI 的 BackgroundTasks 适合响应后执行短小、允许跟随当前进程生命周期的工作，例如写非关键审计日志。耗时模型任务、必须重试的邮件或不可丢失的数据处理，应写入持久化队列并由独立 Worker 消费。

短小且丢失可接受
└─ 进程内 BackgroundTasks

耗时、需重试、需查询状态或不能丢
└─ 持久化任务记录 / 消息队列
   └─ 独立 Worker
      └─ Checkpoint、幂等与重试策略

高频面试题与回答

1. FastAPI 的依赖注入适合放什么？参考答案

适合请求级资源和可复用边界逻辑，例如数据库 Session、当前身份、权限检查和 Service 组装。业务决策仍放在 Service；全应用连接池由 lifespan 管理；不应把所有逻辑都变成层层 Depends。

2. 认证和授权有什么区别？参考答案

认证确认调用方是谁，授权判断这个身份能否执行当前动作。即使令牌有效，也不能直接访问任意资源；查询应加入用户或租户条件，敏感动作还要检查角色和资源关系。

3. 流式响应中途失败为什么不能再返回 500？参考答案

响应头和状态码通常在流开始时已经发送，之后只能继续写响应体或断开连接。因此流式协议要定义错误事件、完成事件和重连语义，并在客户端断开时取消下游任务。

4. FastAPI 启动多个 Worker 后能共享内存状态吗？参考答案

不能把它当成共享状态。多个 Worker 通常是独立进程，各自拥有事件循环、连接池和进程内变量；某个 Worker 修改的全局字典不会自动同步给其他 Worker。跨请求、跨实例的一致状态应放进数据库、Redis 或消息系统。

接下来学什么

下一篇学习 pytest、替身与分层测试，用可控依赖验证路由、Service 和数据库边界。

参考资料

• FastAPI 依赖注入 (opens new window)
• FastAPI 使用 yield 的依赖 (opens new window)
• FastAPI lifespan (opens new window)
• FastAPI 安全教程 (opens new window)
• FastAPI StreamingResponse (opens new window)