FastAPI中如何获取Bearer Token并实现JWT令牌自动刷新

核心思路 #

基于原生python-jose库实现JWT逻辑，自定义FastAPI依赖类完成令牌验证、数据返回及自动刷新，完全规避第三方库依赖冲突问题。

步骤1：安装依赖 #

pip install fastapi python-jose[cryptography] python-multipart

步骤2：基础配置与工具函数 #

from datetime import datetime, timedelta
from fastapi import FastAPI, Depends, HTTPException, status, Request, Response
from fastapi.security import OAuth2PasswordRequestForm
from jose import JWTError, jwt
from pydantic import BaseModel

# 核心配置（生产环境建议用环境变量加载）
SECRET_KEY = "your-secret-key-here"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 15
REFRESH_TOKEN_EXPIRE_DAYS = 7

# 用户数据模型
class TokenData(BaseModel):
    username: str | None = None

# JWT工具函数
def create_access_token(data: dict, expires_delta: timedelta | None = None):
    to_encode = data.copy()
    expire = datetime.utcnow() + (expires_delta or timedelta(minutes=15))
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

def create_refresh_token(data: dict):
    expire = datetime.utcnow() + timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS)
    to_encode = data.copy()
    to_encode.update({"exp": expire, "type": "refresh"})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

def verify_token(token: str):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="无法验证凭证",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username = payload.get("sub")
        if not username:
            raise credentials_exception
        return payload, TokenData(username=username)
    except JWTError:
        raise credentials_exception

步骤3：自定义JWT依赖类（含自动刷新） #

class JWTBearer:
    def __call__(self, request: Request, response: Response):
        # 提取Bearer令牌
        auth_header = request.headers.get("Authorization")
        if not auth_header or not auth_header.startswith("Bearer "):
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="未认证",
                headers={"WWW-Authenticate": "Bearer"},
            )
        token = auth_header.split(" ")[1]
        
        # 验证令牌并解析数据
        payload, token_data = verify_token(token)
        
        # 自动刷新逻辑：当令牌剩余有效期小于5分钟时生成新令牌
        current_timestamp = datetime.utcnow().timestamp()
        if payload["exp"] - current_timestamp < 300:
            new_access_token = create_access_token(data={"sub": token_data.username})
            # 将新令牌放入响应头，供前端更新本地存储
            response.headers["X-New-Access-Token"] = new_access_token
        
        # 返回令牌、payload、用户数据，供路由直接使用
        return token, payload, token_data

步骤4：路由实现 #

app = FastAPI()

# 模拟用户验证逻辑（替换为你的数据库查询逻辑）
def authenticate_user(username: str, password: str):
    if username == "test" and password == "test123":
        return {"username": username}
    return None

# 登录接口：获取初始令牌
@app.post("/login")
def login(form_data: OAuth2PasswordRequestForm = Depends()):
    user = authenticate_user(form_data.username, form_data.password)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="用户名或密码错误",
            headers={"WWW-Authenticate": "Bearer"},
        )
    access_token = create_access_token(
        data={"sub": user["username"]},
        expires_delta=timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    )
    refresh_token = create_refresh_token(data={"sub": user["username"]})
    return {
        "access_token": access_token,
        "refresh_token": refresh_token,
        "token_type": "bearer"
    }

# 受保护接口：直接获取令牌及payload
@app.get("/protected/email-task")
def run_email_task(token_info: tuple = Depends(JWTBearer())):
    token, payload, user_data = token_info
    return {
        "message": "邮件任务执行成功",
        "submitted_token": token,
        "token_payload": payload,
        "operator": user_data.username
    }

# 手动刷新令牌接口（作为自动刷新的备选）
@app.post("/refresh-token")
def refresh_token(refresh_token: str):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="刷新令牌无效",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(refresh_token, SECRET_KEY, algorithms=[ALGORITHM])
        if payload.get("type") != "refresh":
            raise credentials_exception
        username = payload.get("sub")
        if not username:
            raise credentials_exception
        new_access_token = create_access_token(data={"sub": username})
        return {"access_token": new_access_token, "token_type": "bearer"}
    except JWTError:
        raise credentials_exception

关键说明 #

1. 令牌获取规范：通过自定义依赖类JWTBearer统一处理令牌提取与验证，路由只需通过Depends(JWTBearer())即可直接获取令牌、payload和用户数据，无需手动操作Request头。
2. 自动刷新逻辑：在依赖类中检查令牌有效期，当剩余时间不足5分钟时自动生成新的access token并放入响应头X-New-Access-Token，前端可监听该头更新本地令牌。
3. 安全保障：refresh令牌单独生成并标记类型，避免与access令牌混淆；生产环境需将SECRET_KEY通过环境变量注入，禁止硬编码。

内容的提问来源于stack exchange，提问作者Federico Frontini