---
id: python.fastapi.deps-auth-errors
title: 依赖注入、鉴权、中间件和统一错误处理
type: lesson
domain: python
stack: [fastapi]
tags: [Depends, 鉴权, JWT, 中间件, 异常处理器, 多租户]
level: intermediate
prerequisites: [python.fastapi.routing-validation]
est_hours: 3
related: [python.fastapi.streaming-testing, backend.gin.middleware-auth, python.fastapi.interview]
week: [4, 5]
order: 2
---

# 依赖注入、鉴权、中间件和统一错误处理

## 解释说明

**依赖注入 `Depends`**：把“获取当前用户”“获取数据库会话”“获取模型客户端”写成函数，路由参数里声明 `Depends(函数)`，FastAPI 自动调用并把结果传进来。

- 依赖可以嵌套（获取当前用户 → 依赖解析 token → 依赖读取配置）。
- 带 `yield` 的依赖可以做资源清理（请求结束后关闭数据库会话）。
- 同一个请求里同一个依赖默认只执行一次（有缓存）。
- **测试时用 `app.dependency_overrides[原依赖] = 假实现`** 替换，这是 FastAPI 最好用的测试手段。

**鉴权**：常见两种——给服务调用方的 API Key（请求头 `X-API-Key`），给用户的 JWT（`Authorization: Bearer ...`）。写成依赖：校验失败抛 `HTTPException(401)`；权限不足抛 403。多租户系统里，依赖返回“当前租户”，后续查询都带上租户条件。

**中间件**：每个请求都要经过的逻辑——请求 ID、耗时日志、CORS。`@app.middleware("http")` 或添加 Starlette 中间件类。业务相关的校验（鉴权、权限）用依赖更合适，因为能精确控制作用于哪些路由。

**统一错误处理**：`@app.exception_handler(异常类型)`。

- `HTTPException` → 统一格式 `{"code", "message", "request_id"}`；
- `RequestValidationError` → 把 422 转成统一格式，列出字段和原因；
- 兜底 `Exception` → 记录完整日志，返回 500 和通用提示，**不返回内部细节**。

和 Gin 对照：`Depends` 类似在中间件里 `c.Set("user", ...)` 然后在 handler 里取，但它有类型、按路由声明、可以在测试中替换。

## 案例

AI 服务给多个业务系统调用：每个系统一个 API Key，对应一个租户；每个请求日志都带 `request_id` 和租户；参数错误返回统一格式让调用方能程序化处理；内部异常（数据库连接串、堆栈）只进日志。

## 代码示例

```python
from collections.abc import Iterator
from typing import Annotated

from fastapi import Depends, FastAPI, Header, HTTPException, Request
from fastapi.responses import JSONResponse
from fastapi.testclient import TestClient


class Session:                                   # 模拟数据库会话
    def __init__(self) -> None:
        self.closed = False

    def query_docs(self, tenant: str) -> list[str]:
        return [f"{tenant} 的文档 {i}" for i in range(2)]


opened: list[Session] = []


def get_session() -> Iterator[Session]:          # 带 yield 的依赖：请求结束后清理
    s = Session()
    opened.append(s)
    try:
        yield s
    finally:
        s.closed = True


KEYS = {"key-a": "tenant-a", "key-b": "tenant-b"}


def current_tenant(x_api_key: Annotated[str | None, Header()] = None) -> str:
    if not x_api_key or x_api_key not in KEYS:
        raise HTTPException(status_code=401, detail="API Key 无效")
    return KEYS[x_api_key]


def require_admin(tenant: Annotated[str, Depends(current_tenant)]) -> str:   # 依赖嵌套
    if tenant != "tenant-a":
        raise HTTPException(status_code=403, detail="没有管理权限")
    return tenant


app = FastAPI()


@app.exception_handler(HTTPException)
async def http_error(request: Request, exc: HTTPException):
    return JSONResponse({"code": exc.status_code, "message": exc.detail}, status_code=exc.status_code)


@app.get("/documents")
def list_docs(tenant: Annotated[str, Depends(current_tenant)],
              db: Annotated[Session, Depends(get_session)]) -> list[str]:
    return db.query_docs(tenant)                 # 查询一律带租户条件


@app.delete("/cache")
def clear_cache(tenant: Annotated[str, Depends(require_admin)]) -> dict:
    return {"cleared_by": tenant}


c = TestClient(app)
print(c.get("/documents").json())                                        # {'code': 401, 'message': 'API Key 无效'}
print(c.get("/documents", headers={"X-API-Key": "key-b"}).json())        # ['tenant-b 的文档 0', 'tenant-b 的文档 1']
print(opened[-1].closed)                                            # True：请求结束后会话已关闭
print(c.delete("/cache", headers={"X-API-Key": "key-b"}).status_code)  # 403

# 测试里替换依赖：不需要真实的 API Key 和数据库
app.dependency_overrides[current_tenant] = lambda: "tenant-test"
print(c.get("/documents").json())                                        # ['tenant-test 的文档 0', ...]
app.dependency_overrides.clear()
```

请求 ID 中间件、422 统一格式、500 兜底的完整写法见 [examples/chat-api/app.py](examples/chat-api/app.py)。

## 易错点

- **鉴权写在中间件里对所有路径生效**：健康检查、文档页面也被拦住；用依赖按路由声明更精确（或在路由器级别 `dependencies=[Depends(...)]`）。
- **多租户只在入口校验，查询时忘了带租户条件**：越权读到别的租户数据。把租户作为数据访问层的必传参数。
- **兜底异常处理器把 `str(exc)` 返回给调用方**：泄露数据库地址、内部路径。
- **测试后不清理 `dependency_overrides`**：影响其他测试。
- **API Key 明文存储、用 `==` 比较**：存哈希，用 `secrets.compare_digest` 比较以防时序攻击。
- **自定义路由用了 `/docs`、`/redoc`、`/openapi.json`**：和 FastAPI 自带的文档页面冲突（本课示例最初就踩了这个坑，`GET /docs` 返回的是 Swagger 页面）。换路径，或者 `FastAPI(docs_url=None)` 关闭/改名自带文档——生产环境通常也应该关闭或加鉴权。
- **在依赖里做很重的操作**（每个请求都查一次配置中心）：考虑缓存或在应用启动时加载。

## 面试常问

见 [FastAPI 面试题](interview.md)：Depends 的原理和用途、中间件和依赖的区别、怎么统一错误格式、怎么做多租户隔离。