135 lines
6.2 KiB
Markdown
135 lines
6.2 KiB
Markdown
---
|
||
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 的原理和用途、中间件和依赖的区别、怎么统一错误格式、怎么做多租户隔离。
|