前四篇把「能跑」讲完了,这篇讲「能上线」。一个对外服务的硬指标是:谁能调、安不安全、跑得稳不稳定、改了怎么验证。对应到 FastAPI 就是四块:鉴权、跨域与中间件、部署、测试。
认证:OAuth2 密码流 + JWT
FastAPI 内置了 OAuth2 的工具,配合 JWT 做无状态鉴权最常见。
密码哈希
明文存密码是事故。用 passlib 的 bcrypt:
from passlib.context import CryptContext pwd = CryptContext(schemes=["bcrypt"], deprecated="auto") hashed = pwd.hash("明文密码") pwd.verify("明文密码", hashed) # True
注册时存 hashed,登录时 pwd.verify(提交密码, 库里hash) 比对。永远不存明文。
签发与校验 JWT
用 python-jose 签发 token:
from datetime import datetime, timedelta, timezone from jose import jwt SECRET = "你的随机密钥" ALGO = "HS256" def make_token(sub: str) -> str: exp = datetime.now(timezone.utc) + timedelta(hours=12) return jwt.encode({"sub": sub, "exp": exp}, SECRET, algorithm=ALGO) def parse_token(t: str) -> str | None: try: return jwt.decode(t, SECRET, algorithms=[ALGO])["sub"] except Exception: return None
接成依赖
用 OAuth2PasswordBearer 告诉 FastAPI 去哪找 token,再写成获取当前用户的依赖(呼应第 2 篇):
from fastapi import Depends, HTTPException from fastapi.security import OAuth2PasswordBearer oauth2 = OAuth2PasswordBearer(tokenUrl="login") def get_current_user(token: str = Depends(oauth2)): sub = parse_token(token) if not sub: raise HTTPException(status_code=401, detail="无效或过期的 token") return {"username": sub} @app.get("/me") async def me(user: dict = Depends(get_current_user)): return user
tokenUrl="login" 还会让 /docs 自动出现「Authorize」按钮,填用户名密码换 token,方便联调。
跨域:CORS
前端和 API 不同源时必须开 CORS,否则浏览器拦截:
from fastapi.middleware.cors import CORSMiddleware app.add_middleware( CORSMiddleware, allow_origins=["https://www.wxbuluo.com"], # 生产写具体域名,别用 ["*"] allow_credentials=True, allow_methods=["*"], allow_headers=["*"], )
allow_origins=["*"] 配合 allow_credentials=True 在规范里是不允许的,生产环境务必列具体前端域名。
自定义中间件
中间件在请求进入路由前、响应返回后拦截,适合做计时、日志、请求 ID:
import time from fastapi import Request @app.middleware("http") async def add_timing(request: Request, call_next): start = time.perf_counter() response = await call_next(request) # 把请求交给后续处理 cost = (time.perf_counter() - start) * 1000 response.headers["X-Process-Ms"] = f"{cost:.1f}" return response
await call_next(request) 是关键——它驱动后续链路,返回的就是最终响应,你在前后都能插入逻辑。
后台任务:BackgroundTasks
不要在响应里做慢操作(发邮件、写日志、调第三方)。用 BackgroundTasks 把活丢到响应之后跑:
from fastapi import BackgroundTasks def send_welcome_email(to: str): ... # 慢操作 @app.post("/signup") async def signup(background: BackgroundTasks, email: str): background.add_task(send_welcome_email, email) return {"ok": True} # 先立即返回,邮件在响应后异步发
注意 BackgroundTasks 跑在同一个进程里,进程重启会丢任务。真要可靠投递,用消息队列(Celery / Redis Stream),别用这个顶替。
部署:gunicorn + uvicorn workers
开发用 uvicorn main:app --reload,生产要上多 worker 吃满多核:
pip install gunicorn gunicorn main:app -k uvicorn.workers.UvicornWorker -w 4 -b 0.0.0.0:8000
-w 4 起 4 个 worker 进程。worker 数 = CPU 核数 是常用起点;因为 async 单进程已经能扛高并发 IO,CPU 密集时再加。前面 nginx 反代把 80/443 的流量转给 8000 即可(你博客服务器就是这套架构)。
最小 Dockerfile:
FROM python:3.12-slim WORKDIR /app COPY requirements.txt . RUN pip install -r requirements.txt COPY . . CMD ["gunicorn", "main:app", "-k", "uvicorn.workers.UvicornWorker", "-w", "4", "-b", "0.0.0.0:8000"]
测试:TestClient + 依赖覆盖
FastAPI 用 Starlette 的 TestClient(底层 requests)做同步风格测试,pytest 直接跑:
from fastapi.testclient import TestClient client = TestClient(app) def test_read_root(): r = client.get("/") assert r.status_code == 200
真正有用的是依赖覆盖——把依赖数据库的那个 get_db 换成测试库或假数据,不碰真实存储:
def fake_db(): yield InMemoryDB() # 测试用假 session app.dependency_overrides[get_db] = fake_db def test_list_posts(): r = client.get("/posts/") assert r.status_code == 200
dependency_overrides 是 FastAPI 测试的核心武器:任何 Depends 都能被替换,单元测试不必起真数据库。
五篇回顾
这个系列从定位讲到生产,脉络是:
- 是什么:类型注解驱动 + 原生异步 + 自动文档,定位 API 服务
- 路由与依赖:参数从声明自动解析,Depends 把横切逻辑做成层级复用
- Pydantic:用类型注解把「数据长什么样、哪些合法」锁死,请求/响应模型分离
- 异步数据库:SQLAlchemy 2.0 async 写法 + 依赖注入 session + 同步异步边界
- 生产化:JWT 鉴权、CORS、中间件、后台任务、gunicorn 部署、pytest 依赖覆盖
FastAPI 的价值不在「多」,而在把校验、文档、异步、依赖注入这几件高频又易错的事,做成框架默认能力。把这套跑顺,你写 API 的样板代码会少一大半。