FastAPI与React匿名用户会话管理：基于JWT的实践指南

pip install "fastapi[all]" python-jose[cryptography] passlib[bcrypt]

# app/core/security.py
from datetime import datetime, timedelta
from typing import Optional

from jose import jwt, JWTError
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from pydantic import BaseModel

# JWT配置
SECRET_KEY = "your_super_secret_key" # 生产环境中请使用更安全的密钥，并从环境变量获取
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30 # 匿名会话令牌有效期

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token") # tokenUrl可以是任何你定义的登录路径

class TokenData(BaseModel):
    username: Optional[str] = None

def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

async def get_current_user(token: str = Depends(oauth2_scheme)):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
        token_data = TokenData(username=username)
    except JWTError:
        raise credentials_exception
    return token_data.username # 返回用户名，这里就是我们的匿名ID

# app/main.py
from fastapi import FastAPI, Response
from pydantic import BaseModel
import uuid
from app.core.security import create_access_token, get_current_user

app = FastAPI()

class Token(BaseModel):
    access_token: str
    token_type: str

@app.post("/anonymous-login", response_model=Token)
async def anonymous_login():
    """
    为首次访问用户生成匿名ID并返回JWT。
    """
    anonymous_id = f"anonymous_{uuid.uuid4().hex}" # 生成一个唯一的匿名ID
    access_token_expires = timedelta(minutes=30) # 匿名令牌有效期
    access_token = create_access_token(
        data={"sub": anonymous_id}, expires_delta=access_token_expires
    )
    # 可以在这里将 anonymous_id 存储到数据库中，以便后续跟踪其行为
    print(f"Generated anonymous ID: {anonymous_id}")
    return {"access_token": access_token, "token_type": "bearer"}

# app/main.py (续)
from fastapi import Depends

@app.get("/user/me")
async def read_current_user(current_user: str = Depends(get_current_user)):
    """
    示例：获取当前用户的ID（可能是匿名ID或真实用户ID）。
    """
    return {"current_user_id": current_user}

@app.get("/anonymous-data")
async def get_anonymous_data(current_user_id: str = Depends(get_current_user)):
    """
    示例：根据匿名用户ID获取其历史数据。
    """
    if not current_user_id.startswith("anonymous_"):
        # 实际应用中，这里可能需要处理非匿名用户的逻辑
        return {"message": f"Welcome, registered user {current_user_id}!"}

    # 从数据库中根据 current_user_id 查询该匿名用户的历史行为
    # 假设这里有一个模拟的数据库查询
    mock_db = {
        "anonymous_12345": {"last_viewed_product": "Product A", "cart_items": 2},
        "anonymous_67890": {"last_viewed_product": "Product B", "cart_items": 0},
    }
    user_data = mock_db.get(current_user_id, {"message": "No data found for this anonymous user."})
    return {"anonymous_id": current_user_id, "data": user_data}

// src/api.js (示例：使用Axios)
import axios from 'axios';

const API_BASE_URL = 'http://localhost:8000'; // 你的FastAPI后端地址

const api = axios.create({
  baseURL: API_BASE_URL,
  headers: {
    'Content-Type': 'application/json',
  },
});

// 请求拦截器：在每次请求前添加Authorization头
api.interceptors.request.use(
  (config) => {
    const token = localStorage.getItem('anonymous_access_token');
    if (token) {
      config.headers.Authorization = `Bearer ${token}`;
    }
    return config;
  },
  (error) => {
    return Promise.reject(error);
  }
);

export const getAnonymousToken = async () => {
  try {
    const response = await api.post('/anonymous-login');
    const { access_token } = response.data;
    localStorage.setItem('anonymous_access_token', access_token);
    return access_token;
  } catch (error) {
    console.error('Error getting anonymous token:', error);
    throw error;
  }
};

export const getAnonymousData = async () => {
  try {
    const response = await api.get('/anonymous-data');
    return response.data;
  } catch (error) {
    console.error('Error fetching anonymous data:', error);
    throw error;
  }
};

export const getUserMe = async () => {
  try {
    const response = await api.get('/user/me');
    return response.data;
  } catch (error) {
    console.error('Error fetching user me:', error);
    throw error;
  }
};

// src/App.js (示例React组件)
import React, { useEffect, useState } from 'react';
import { getAnonymousToken, getAnonymousData, getUserMe } from './api';

function App() {
  const [anonymousId, setAnonymousId] = useState(null);
  const [userData, setUserData] = useState(null);

  useEffect(() => {
    const initializeAnonymousSession = async () => {
      let token = localStorage.getItem('anonymous_access_token');
      if (!token) {
        // 如果没有令牌，则获取一个新的匿名令牌
        token = await getAnonymousToken();
      }
      // 无论是否新获取，都尝试获取用户数据
      try {
        const userMeResponse = await getUserMe();
        setAnonymousId(userMeResponse.current_user_id);
        const dataResponse = await getAnonymousData();
        setUserData(dataResponse);
      } catch (error) {
        console.error('Failed to fetch user data with token:', error);
        // 如果令牌失效，可以尝试重新获取
        localStorage.removeItem('anonymous_access_token');
        // 刷新页面或提示用户
      }
    };

    initializeAnonymousSession();
  }, []);

  return (
    <div style={{ padding: '20px' }}>
      <h1>匿名用户会话示例</h1>
      <p>当前用户ID: {anonymousId || '加载中...'}</p>
      {userData && (
        <div>
          <h2>用户数据:</h2>
          <pre>{JSON.stringify(userData, null, 2)}</pre>
        </div>
      )}
      <p>刷新页面，匿名ID通常会保持不变，因为令牌存储在LocalStorage中。</p>
    </div>
  );
}

export default App;