FastAPI-Day5

本篇由AI生成

前置知识：HTTP 的无状态性

HTTP 协议天生是无状态的——每次请求之间相互独立，服务器认不出”这是同一个人发来的请求”。

Cookie 和 Session 就是用来给 HTTP 补上”记忆”的两种方式：

方式	数据存哪	特点
Cookie	浏览器	每次请求自动带上，大小有限(4KB)
Session	服务端	安全，可存大量数据，依赖 Cookie

Cookie 是服务器通过响应头 Set-Cookie 下发给浏览器的一小段文本（键值对），浏览器会将其保存并在后续请求中通过请求头 Cookie 带回来。

整个过程如下：

客户端                         服务端
  |                             |
  |---- 请求 /login ----------->|
  |                             |  Set-Cookie: session_id=abc123
  |<--- 响应 + Set-Cookie ------|
  |                             |
  |---- 请求 /profile --------->|
  |    Cookie: session_id=abc123|
  |                             |  通过 Cookie 识别出用户
  |<--- 返回用户信息 -----------|

使用 Response.set_cookie() 方法：

from fastapi import FastAPI, Response

app = FastAPI()

@app.get("/set-cookie")
async def set_cookie(response: Response):
    response.set_cookie(
        key="username",          # Cookie 名称
        value="豆棍度子"         # Cookie 值
    )
    return {"message": "Cookie 已设置"}

set_cookie() 的常用参数：

参数	类型	说明
key	`str`	Cookie 名称，必填
value	`str`	Cookie 值，必填
max_age	`int`	存活秒数，不设则为会话 Cookie（关闭浏览器即失效）
expires	`datetime`	过期时间（与 max_age 二选一）
path	`str`	Cookie 生效路径，默认 `/`（全站有效）
domain	`str`	Cookie 生效域名，默认当前域名
secure	`bool`	仅 HTTPS 传输
httponly	`bool`	禁止 JavaScript 访问（防 XSS）
samesite	`str`	限制跨站发送：`"lax"` / `"strict"` / `"none"`

设置了这些参数后的完整示例：

@app.get("/set-cookie-secure")
async def set_secure_cookie(response: Response):
    response.set_cookie(
        key="session_id",
        value="abc123xyz",
        max_age=60 * 60 * 24,        # 24 小时后过期
        httponly=True,               # JS 无法读取
        samesite="lax",              # 同站请求才发送
        secure=True,                 # 仅 HTTPS（生产环境）
    )
    return {"message": "安全 Cookie 已设置"}

用 Cookie() 依赖注入，参数名与 Cookie 的 key 对应：

from fastapi import Cookie

@app.get("/get-cookie")
async def get_cookie(username: str | None = Cookie(None)):
    if username:
        return {"username": username}
    return {"message": "没有找到 Cookie"}

Cookie(None) 表示默认为 None，这样即使用户没有该 Cookie 也不会报错。

也可以一次性获取所有 Cookie：

from fastapi import Request

@app.get("/all-cookies")
async def get_all_cookies(request: Request):
    cookies = request.cookies          # 字典形式返回所有 Cookie
    return cookies

使用 Response.delete_cookie()：

@app.get("/delete-cookie")
async def delete_cookie(response: Response):
    response.delete_cookie(key="username")
    return {"message": "Cookie 已删除"}

delete_cookie() 本质上是将 Cookie 的 max_age 设为 0，使浏览器立即将其过期。

from fastapi import FastAPI, Response, Cookie

app = FastAPI()

# 支持的 theme
VALID_THEMES = {"light", "dark", "auto"}

@app.get("/")
async def index(theme: str | None = Cookie(None)):
    if theme and theme in VALID_THEMES:
        return {"message": f"当前主题：{theme}"}
    return {"message": "请先设置主题"}

@app.get("/set-theme/{theme_name}")
async def set_theme(theme_name: str, response: Response):
    if theme_name not in VALID_THEMES:
        return {"error": f"不支持的主题，可选：{VALID_THEMES}"}

    response.set_cookie(
        key="theme",
        value=theme_name,
        max_age=60 * 60 * 24 * 30,    # 30 天
        httponly=True,
    )
    return {"message": f"主题已切换为：{theme_name}"}

Session

什么是 Session

Session（会话）将数据存储在服务端，只给浏览器发一个唯一的 Session ID（通过 Cookie 保存）。浏览器每次请求带上这个 ID，服务端就能根据 ID 找到对应用户的数据。

                           服务端
客户端              ┌──────────────────┐
  |                 │  Session 存储    │
  |                 │                  │
  |  Cookie:        │  abc123 → {      │
  |  session=abc123 │    username: "豆"│
  |──── /profile ──>│    cart: [...]   │
  |                 │    login: true   │
  |<── 用户信息 ──── │ }                │
  |                 └──────────────────┘

方式一：SessionMiddleware（最简单）

Starlette 内置了 SessionMiddleware，它把 Session 数据加密后存在 Cookie 中（服务端不存数据），适合小项目：

from fastapi import FastAPI, Request
from starlette.middleware.sessions import SessionMiddleware

app = FastAPI()

# 添加 Session 中间件
app.add_middleware(
    SessionMiddleware,
    secret_key="请替换成随机密钥",      # 加密密钥，务必替换
    session_cookie="session",           # Cookie 名称，默认 session
    max_age=60 * 60 * 24 * 7,           # 过期时间 7 天，默认 2 周
    same_site="lax",                    # 同站策略
    https_only=False,                   # 生产环境应设为 True
)

配置好后，直接在 request.session 中读写（行为类似字典）：

@app.get("/login")
async def login(request: Request):
    request.session["username"] = "豆棍度子"    # 写入 session
    request.session["is_login"] = True
    return {"message": "登录成功"}

@app.get("/profile")
async def profile(request: Request):
    username = request.session.get("username")
    if not username:
        return {"error": "未登录"}

    return {
        "username": username,
        "is_login": request.session.get("is_login"),
    }

@app.get("/logout")
async def logout(request: Request):
    request.session.clear()                    # 清空所有 session 数据
    return {"message": "已退出登录"}

SessionMiddleware 基于 itsdangerous 库对数据进行签名加密，数据本身存储在 Cookie 中（客户端 session），所以不要存敏感信息（如密码、信用卡号）。

方式二：服务端 Session（更安全）

对于需要存储敏感数据或大量数据的场景，应使用服务端 Session——Session ID 存在 Cookie 中，实际数据存在 Redis 或数据库中。

基于 Redis 的 Session

import json
import uuid
from datetime import datetime, timedelta

import redis.asyncio as aioredis
from fastapi import FastAPI, Request, Response, Cookie

app = FastAPI()

# 连接 Redis（本地开发）
redis_client = aioredis.from_url(
    "redis://localhost:6379/0",
    decode_responses=True
)

SESSION_EXPIRE = timedelta(hours=2)    # Session 过期时间

async def create_session(data: dict) -> str:
    """创建 Session，返回 Session ID"""
    session_id = uuid.uuid4().hex       # 生成唯一 ID
    await redis_client.setex(
        f"session:{session_id}",
        int(SESSION_EXPIRE.total_seconds()),
        json.dumps(data)
    )
    return session_id

async def get_session(session_id: str) -> dict | None:
    """根据 Session ID 获取数据"""
    data = await redis_client.get(f"session:{session_id}")
    if data:
        return json.loads(data)
    return None

async def delete_session(session_id: str):
    """删除 Session"""
    await redis_client.delete(f"session:{session_id}")

@app.post("/login")
async def login(response: Response):
    # 模拟用户登录
    user_data = {
        "user_id": 1,
        "username": "豆棍度子",
        "role": "admin",
        "login_time": datetime.now().isoformat()
    }
    session_id = await create_session(user_data)

    # 将 Session ID 下发给浏览器
    response.set_cookie(
        key="session_id",
        value=session_id,
        max_age=int(SESSION_EXPIRE.total_seconds()),
        httponly=True,
        samesite="lax",
    )
    return {"message": "登录成功"}

@app.get("/profile")
async def profile(session_id: str | None = Cookie(None)):
    if not session_id:
        return {"error": "未登录"}

    user_data = await get_session(session_id)
    if not user_data:
        return {"error": "Session 已过期，请重新登录"}

    return {"user": user_data}

@app.post("/logout")
async def logout(response: Response, session_id: str | None = Cookie(None)):
    if session_id:
        await delete_session(session_id)          # 删除 Redis 中的 Session
        response.delete_cookie(key="session_id")  # 删除浏览器中的 Cookie

    return {"message": "已退出登录"}

依赖注入版 Session（推荐）

把获取当前用户的逻辑封装成依赖，路由函数直接使用：

from fastapi import Depends, HTTPException, status
from fastapi import Cookie as CookieDep

# 提取 Session 的依赖
async def get_current_user(
    session_id: str | None = CookieDep(None)
) -> dict:
    if not session_id:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="未登录"
        )

    user_data = await get_session(session_id)
    if not user_data:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Session 已过期"
        )

    return user_data

# 路由直接使用依赖
@app.get("/profile")
async def profile(current_user: dict = Depends(get_current_user)):
    return {"user": current_user}

@app.get("/admin")
async def admin_panel(current_user: dict = Depends(get_current_user)):
    if current_user.get("role") != "admin":
        raise HTTPException(status_code=403, detail="权限不足")

    return {"message": "欢迎管理员", "user": current_user}

Session 方式对比

方式	存储位置	数据量	安全性	需要额外服务	适用场景
SessionMiddleware	Cookie	小	中（加密）	不需要	小项目、原型、个人博客
Redis Session	服务端	大	高	Redis	生产项目、需要共享登录态
数据库 Session	数据库	大	高	数据库	已有数据库不想引入 Redis

常见安全注意事项

response.set_cookie(
    key="session_id",
    value="...",
    httponly=True,      # 防止 XSS 窃取 Cookie
    samesite="lax",     # 防止 CSRF 攻击
    secure=True,        # 仅 HTTPS 传输（生产环境）
)

属性	防护目标	说明
`httponly`	XSS（脚本注入）	JS 无法读取该 Cookie
`samesite`	CSRF（跨站请求伪造）	限制跨站请求携带 Cookie
`secure`	中间人攻击	只在 HTTPS 连接中传输

Session 固定攻击

用户登录后应重新生成 Session ID，防止攻击者诱导用户使用已知的 Session ID 登录：

@app.post("/login")
async def login(response: Response, request: Request):
    # 登录成功后，生成全新的 Session
    old_session_id = request.cookies.get("session_id")
    if old_session_id:
        await delete_session(old_session_id)     # 删除旧 Session

    session_id = await create_session(user_data) # 创建全新的
    response.set_cookie(key="session_id", value=session_id, httponly=True)
    return {"message": "登录成功"}

不要存什么

无论是 Cookie 还是 Session，都不要存储：

❌ 密码明文
❌ 信用卡号
❌ 身份证号
❌ Token / API 密钥（除非加密）

应该只存最基本的用户标识（user_id、username、role），需要更多信息时根据 ID 再查数据库。

对比项	Cookie	Session
存储位置	浏览器	服务端（内存、Redis、数据库）
数据大小	约 4KB	理论无上限
安全性	低（明文可见，可篡改）	高（数据在服务端）
性能	快（服务端无开销）	每次请求需查存储
跨域	有限制（同源策略）	无限制（共享存储即可）
典型用途	记住偏好、跟踪行为	登录态、购物车、敏感数据
FastAPI 实现	`Response.set_cookie()` + `Cookie()`	`SessionMiddleware` 或 Redis 服务端 Session

Cookie 与 Session

前置知识：HTTP 的无状态性

Cookie

什么是 Cookie

在 FastAPI 中设置 Cookie

读取 Cookie

删除 Cookie

Cookie 实战：记录用户偏好