如何在FastAPI OAuth中获取刷新令牌并在所有API中使用

一、创建获取Access/Refresh Token的API端点 #

首先定义请求体模型和响应模型，调用外部API获取双令牌并返回指定格式的响应。注意要确认外部API的授权类型支持返回Refresh Token（部分API的client_credentials模式仅返回Access Token，需查阅外部API文档调整授权类型）。

from pydantic import BaseModel
from fastapi import FastAPI, HTTPException
import requests

app = FastAPI()

# 接收客户端凭证的请求模型
class ClientCredentials(BaseModel):
    client_id: str
    client_secret: str

# 令牌数据模型（匹配外部API返回结构）
class TokenData(BaseModel):
    AccessToken: str
    ExpiresIn: int
    TokenType: str
    RefreshToken: str

# 最终响应模型
class TokenResponse(BaseModel):
    message: str
    data: TokenData

@app.post("/get-tokens", response_model=TokenResponse)
def fetch_tokens(credentials: ClientCredentials):
    # 替换为实际的外部API令牌端点
    external_token_endpoint = "https://external-api.com/oauth/token"
    
    # 外部API所需的请求参数（根据外部文档调整grant_type等字段）
    payload = {
        "client_id": credentials.client_id,
        "client_secret": credentials.client_secret,
        "grant_type": "client_credentials"  # 若该模式不返回Refresh Token，需更换为支持的类型（如authorization_code）
    }

    try:
        # 调用外部API获取令牌
        resp = requests.post(external_token_endpoint, data=payload)
        resp.raise_for_status()
        external_token_data = resp.json()

        # 映射外部数据到自定义响应格式
        token_data = TokenData(
            AccessToken=external_token_data["access_token"],
            ExpiresIn=external_token_data["expires_in"],
            TokenType=external_token_data["token_type"],
            RefreshToken=external_token_data["refresh_token"]
        )

        return TokenResponse(
            message="Validation successful",
            data=token_data
        )
    except requests.exceptions.RequestException as e:
        status_code = resp.status_code if 'resp' in locals() else 500
        raise HTTPException(status_code=status_code, detail=f"令牌获取失败: {str(e)}")

二、在受保护API中使用令牌验证 #

通过FastAPI的依赖项实现令牌的统一验证，确保所有需要授权的API都能自动校验请求头中的AccessToken。

1. 定义令牌验证依赖 #

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer

# 指定令牌获取的端点（即上文的/get-tokens）
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/get-tokens")

def validate_access_token(token: str = Depends(oauth2_scheme)):
    # 此处添加令牌有效性校验逻辑：
    # - 若为JWT令牌，可使用PyJWT解码验证签名、过期时间
    # - 也可调用外部API的令牌验证端点确认有效性
    if not token:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="无效或缺失的令牌",
            headers={"WWW-Authenticate": "Bearer"},
        )
    
    # 示例JWT验证逻辑（需安装pyjwt：pip install pyjwt）
    # import jwt
    # try:
    #     payload = jwt.decode(token, "外部API的公钥/密钥", algorithms=["HS256"])
    # except jwt.PyJWTError:
    #     raise HTTPException(status_code=401, detail="令牌验证失败")
    
    return token

2. 保护API端点 #

在需要授权的API中注入上述依赖即可实现自动验证：

@app.get("/protected-data")
def get_protected_data(valid_token: str = Depends(validate_access_token)):
    # 使用验证后的令牌调用外部受保护资源
    external_protected_endpoint = "https://external-api.com/protected"
    headers = {"Authorization": f"Bearer {valid_token}"}

    try:
        resp = requests.get(external_protected_endpoint, headers=headers)
        resp.raise_for_status()
        return {"message": "受保护资源获取成功", "data": resp.json()}
    except requests.exceptions.RequestException as e:
        status_code = resp.status_code if 'resp' in locals() else 500
        raise HTTPException(status_code=status_code, detail=f"资源获取失败: {str(e)}")

三、进阶：后端自动管理令牌生命周期 #

若需后端自动维护令牌（过期自动刷新），可通过缓存存储Refresh Token，并在Access Token过期时自动调用外部API刷新：

# 生产环境建议使用Redis等分布式缓存，此处用字典模拟
token_cache = {}

def get_cached_tokens(client_id: str):
    return token_cache.get(client_id)

def update_cached_tokens(client_id: str, tokens: TokenData):
    token_cache[client_id] = tokens

# 刷新令牌端点
@app.post("/refresh-token", response_model=TokenResponse)
def refresh_access_token(client_id: str, refresh_token: str):
    external_refresh_endpoint = "https://external-api.com/oauth/refresh"
    payload = {
        "client_id": client_id,
        "refresh_token": refresh_token,
        "grant_type": "refresh_token"
    }

    try:
        resp = requests.post(external_refresh_endpoint, data=payload)
        resp.raise_for_status()
        new_token_data = resp.json()

        updated_token_data = TokenData(
            AccessToken=new_token_data["access_token"],
            ExpiresIn=new_token_data["expires_in"],
            TokenType=new_token_data["token_type"],
            RefreshToken=new_token_data.get("refresh_token", refresh_token)  # 部分API刷新后返回新的Refresh Token
        )

        update_cached_tokens(client_id, updated_token_data)

        return TokenResponse(
            message="令牌刷新成功",
            data=updated_token_data
        )
    except requests.exceptions.RequestException as e:
        status_code = resp.status_code if 'resp' in locals() else 500
        raise HTTPException(status_code=status_code, detail=f"令牌刷新失败: {str(e)}")

关键注意事项 #

• 外部API授权类型确认：若client_credentials模式不返回Refresh Token，需更换为外部API支持的模式（如authorization_code或password），具体参考外部API文档。
• 安全存储：生产环境中禁止用字典缓存令牌，需使用Redis等加密缓存服务，避免敏感信息泄露。
• 严格验证：令牌验证需覆盖签名校验、过期时间检查等，不可仅做存在性判断。
• 错误处理：完善异常捕获逻辑，处理外部API的各种错误状态（如401、403、超时等）。

内容的提问来源于stack exchange，提问作者Vinuta Basavaraj Hiremath