from typing import Annotated
from fastapi import FastAPI, Depends, Header, HTTPException, status
app = FastAPI()
# 1. 编写自定义校验函数
# 这里的参数名会自动对应 HTTP 请求头 (Headers) 里的字段名
def verify_auth(
username: Annotated[str, Header()],
token: Annotated[str, Header()]
):
# 这里写你的校验逻辑,比如查询数据库或缓存
# 假设正确的组合是 admin / secret123
is_valid = (username == "admin" and token == "secret123")
if not is_valid:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="用户名或 Token 错误,认证失败"
)
# 校验通过后,可以返回用户名,方便后续接口使用
return username
# 2. 将校验函数设为全局依赖
# 这样整个 App 下的所有接口都会自动先跑这个函数
app.dependencies = [Depends(verify_auth)]
# --- 以下是接口示例 ---
@app.get("/data")
async def get_data():
# 因为有全局依赖,只有验证通过才会进到这里
return {"message": "认证成功,这是受保护的数据"}
@app.post("/upload")
async def upload_file():
return {"message": "文件上传成功"}
# 如果某个接口需要知道当前是谁在操作,可以单独再接收一下
@app.get("/me")
async def read_current_user(username: Annotated[str, Depends(verify_auth)]):
return {"current_user": username}
为什么这样写最简洁?
Header 自动提取:通过 Header(),FastAPI 会自动去请求头里找 username 和 token 字段,你不需要手动解析。
全局拦截:app.dependencies = [Depends(...)] 这一行代码,直接省去了在每个 def 后面写一遍验证的麻烦。
标准报错:使用 HTTPException 会自动返回标准的 JSON 格式,前端处理非常统一。