FastApi实现每个会话独立维护游戏状态的技术咨询

你当前的实现把游戏状态存在app全局变量里，导致所有用户共享同一个游戏实例。下面针对你的疑问逐一解答：

1. FastAPI是否有原生方案实现该需求？ #

FastAPI本身没有内置的会话管理功能，但它基于Starlette框架，可以直接用Starlette的SessionMiddleware来实现会话跟踪。这个中间件会给每个浏览器会话分配唯一ID并存在Cookie中，结合内存存储（仅开发环境）、Redis或数据库，就能为每个会话保存独立的游戏状态。

2. 如何生成唯一ID，避免页面刷新时重复生成？ #

用SessionMiddleware的话，它会自动处理会话ID的生成与维护：

• 用户首次访问时，中间件会生成唯一的会话ID，通过Cookie返回给浏览器
• 页面刷新或后续请求时，浏览器会自动带上这个Cookie，后端就能识别出是同一个会话，不会重复生成ID

如果要手动实现（不依赖中间件），可以用uuid.uuid4()生成唯一ID，然后将ID存在Cookie中，每次请求时校验Cookie中的ID是否有效，有效就复用，无效才生成新的。

下面是结合SessionMiddleware的代码示例：

首先添加会话中间件，并准备存储会话游戏状态的容器：

from fastapi import FastAPI, Request, Query, RedirectResponse
from fastapi.templating import Jinja2Templates
from starlette.middleware.sessions import SessionMiddleware
import uuid
# 导入你自己的Player、Game类和NO_WINNER常量
from your_game_modules import Player, Game, NO_WINNER

app = FastAPI()
# 生产环境要换成随机生成的安全密钥，不要硬编码
app.add_middleware(SessionMiddleware, secret_key="your-strong-secret-key")
templates = Jinja2Templates(directory="templates")

# 开发环境用内存字典存游戏状态，生产环境建议替换为Redis或数据库
session_game_store = {}

然后修改路由逻辑，为每个会话维护独立游戏实例：

@app.get("/")
def game_index(request: Request):
    # 检查当前会话是否已有游戏ID
    if "game_id" not in request.session:
        # 生成新游戏ID并存入会话
        game_id = str(uuid.uuid4())
        request.session["game_id"] = game_id
        # 初始化该会话的游戏状态
        players = {0: Player(), 1: Player()}
        session_game_store[game_id] = {
            "players": players,
            "game": Game(players),
            "turn": 0,
            "winner": NO_WINNER
        }
    
    # 获取当前会话的游戏数据
    current_game = session_game_store[request.session["game_id"]]
    return templates.TemplateResponse("index.html", {
        "request": request,
        "players": current_game["players"],
        "turn": current_game["turn"] % 2,
        "winner": current_game["winner"]
    })

@app.get("/select/")
def square_selected(request: Request, player: int = Query(ge=0, le=1), square: int = Query(ge=0, le=5)):
    game_id = request.session.get("game_id")
    # 校验会话有效性，无效则跳转回首页初始化
    if not game_id or game_id not in session_game_store:
        return RedirectResponse(url="/")
    
    current_game = session_game_store[game_id]
    # 处理玩家的落子操作
    current_game["winner"], current_game["turn"] = current_game["game"].calculate_move(
        player, square, current_game["turn"]
    )
    return templates.TemplateResponse("index.html", {
        "request": request,
        "players": current_game["players"],
        "turn": current_game["turn"] % 2,
        "winner": current_game["winner"]
    })

3. 如何判断并清理数据库中的旧会话？ #

不同存储方案的清理方式不同：

• 内存存储（开发环境）：给每个游戏状态添加last_active字段，每次请求时更新该字段。然后用定时任务（比如APScheduler）定期遍历字典，删除last_active超过设定阈值（比如30分钟）的条目。
• Redis：存储游戏状态时给键设置过期时间（比如EX 1800，即30分钟），Redis会自动清理过期的会话数据，无需手动处理。
• 关系型数据库：给会话表添加last_active字段，每次请求时更新该字段。然后用定时任务定期执行SQL，删除last_active早于当前时间30分钟的记录。

另外关于WebSocket的补充：如果想避免每次点击重建连接，可以改用WebSocket长连接方案，前端点击格子时通过WebSocket发送落子请求，后端处理后主动推送游戏状态更新，这样不需要刷新页面，体验更流畅，会话跟踪也更可靠。如果坚持用页面刷新的方式，上面的会话方案完全可以满足需求。

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