FastApi匿名用户无Cookie会话实现及GetCurrentUser适配咨询

核心思路是给匿名用户生成临时身份令牌，替代登录用户的访问令牌来标识会话——这完全契合官方GetCurrentUser方案的设计逻辑：用令牌关联用户/会话身份，匿名场景只是把"登录用户令牌"换成"临时会话令牌"。

具体实现步骤&代码示例 #

1. 基础结构定义

先定义用户模型、会话存储（生产环境建议用Redis替代内存），以及自定义的OAuth2令牌处理类：

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from pydantic import BaseModel
from uuid import uuid4
from typing import Dict, Optional

# 内存会话存储（生产环境替换为Redis）
session_store: Dict[str, Dict] = {}

# 登录用户模型
class User(BaseModel):
    id: str
    username: str

# 匿名用户模型，标识会话身份
class AnonymousUser(BaseModel):
    id: str
    is_anonymous: bool = True

# 自定义OAuth2依赖：无令牌时自动生成匿名会话令牌
class AnonymousOAuth2Bearer(OAuth2PasswordBearer):
    async def __call__(self, request):
        try:
            # 优先尝试获取登录用户的访问令牌
            token = await super().__call__(request)
            return token
        except HTTPException:
            # 无有效登录令牌时，生成匿名会话令牌并初始化会话数据
            anonymous_token = str(uuid4())
            session_store[anonymous_token] = {"data": {}}
            return anonymous_token

oauth2_scheme = AnonymousOAuth2Bearer(tokenUrl="token")

2. 改造GetCurrentUser依赖

在依赖中统一处理登录用户和匿名会话，返回对应的身份实例：

async def get_current_user(token: str = Depends(oauth2_scheme)):
    # 模拟验证登录用户令牌的逻辑（实际从数据库/缓存校验）
    logged_user = await validate_login_token(token)
    if logged_user:
        return logged_user
    
    # 校验是否为已存在的匿名会话令牌
    if token in session_store:
        return AnonymousUser(id=token)
    
    # 无效令牌时重新生成匿名会话（可选，也可抛出错误）
    new_token = str(uuid4())
    session_store[new_token] = {"data": {}}
    return AnonymousUser(id=new_token)

# 模拟登录令牌验证函数
async def validate_login_token(token: str) -> Optional[User]:
    # 实际业务中根据令牌查询用户，这里返回None模拟无登录用户
    return None

3. 接口中使用会话数据

通过get_current_user依赖获取身份，直接操作对应会话的数据：

app = FastAPI()

# 获取当前会话数据
@app.get("/session-data")
async def get_session_data(current_user: User | AnonymousUser = Depends(get_current_user)):
    if isinstance(current_user, AnonymousUser):
        session_data = session_store[current_user.id]["data"]
        return {"is_anonymous": True, "session_data": session_data}
    return {"is_anonymous": False, "user_info": current_user.dict()}

# 更新会话数据
@app.post("/update-session-data")
async def update_session_data(key: str, value: str, current_user: User | AnonymousUser = Depends(get_current_user)):
    if isinstance(current_user, AnonymousUser):
        session = session_store[current_user.id]
        session["data"][key] = value
        return {"message": "会话数据更新成功", "updated_data": session["data"]}
    return {"message": "登录用户无需维护匿名会话数据"}

关键细节说明 #

• 客户端适配：第一次请求无令牌时，服务端会自动生成匿名令牌，需在响应中告知客户端（比如在响应头X-Anonymous-Token返回，或在响应体中携带），后续请求客户端要在Authorization头中带上Bearer <匿名令牌>。
• 会话过期：生产环境要给会话数据添加过期时间，定时清理无效会话，避免存储资源浪费。
• 安全性：务必用HTTPS传输令牌，防止匿名会话被劫持。

内容的提问来源于stack exchange，提问作者Henry Thornton