FastAPI – Session 與 Cookie 管理

整合 Redis / Database Session Backend

---

簡介

在 Web 應用程式中，使用者的登入狀態、購物車資料、臨時設定等資訊往往需要跨請求保存。
傳統的做法是把這些資料寫入瀏覽器的 Cookie，但 Cookie 大小受限且安全性較低，容易被竊取或篡改。

為了在 FastAPI 中提供更安全、彈性的會話管理，我們常會把 Session 資料儲存在 Redis 或 關聯式資料庫，再透過一個唯一的 Session ID（存於 Cookie）來對應。
本篇文章將說明為什麼要這樣做、核心概念、實作方式以及常見陷阱，幫助你在 FastAPI 專案中快速搭建可靠的 Session 後端。

---

核心概念

1️⃣ Session 與 Cookie 的角色分離

• Cookie: 只負責保存一個短小的 Session ID（通常是隨機字串），並在每次請求時自動送回伺服器。
• Session Store: 真正存放使用者資料的地方，可能是 Redis、PostgreSQL、MySQL…等。

把資料搬到 Server 端可以避免客戶端直接看到敏感資訊，也讓資料容量不受 Cookie 限制。

---

2️⃣ 為什麼選擇 Redis？

特性	優點
記憶體快取	讀寫毫秒級，適合高併發的會話存取
自動過期	EXPIRE 可直接設定 Session TTL，省去額外清理機制
分散式支援	可在多台機器間共享 Session，配合 Docker/K8s 更容易水平擴展

---

3️⃣ 為什麼也可以使用關聯式資料庫？

• 持久化需求：某些業務需要 Session 資料在系統重啟後仍保留（例如長時間的購物車）。
• 一致性：若已有資料庫架構，直接新增 session 表即可，減少額外基礎設施。
• 複雜查詢：可以利用 SQL 直接分析 Session 相關統計（如活躍使用者數）。

---

4️⃣ FastAPI 整合方式概覽

1. 建立 Session Middleware：攔截請求、讀取 Cookie 中的 Session ID，從 Store 取出資料並掛載到 request.state.session。
2. 在路由內使用：直接讀寫 request.state.session，中間件會在回應前自動寫回 Store 並設定 Cookie。
3. 設定安全屬性：HttpOnly, Secure, SameSite 等，確保 Session ID 不被惡意腳本竊取。

---

程式碼範例

以下示範 3 種不同的實作，從最簡單的內存儲存，到 Redis、再到 PostgreSQL。每段程式碼皆附有說明註解。

4.1 基礎 In‑Memory Session（僅示範概念）

# file: app/middleware/in_memory_session.py
from fastapi import Request, Response
from starlette.middleware.base import BaseHTTPMiddleware
import uuid
import time

# 全域字典作為簡易 Store（僅開發環境使用）
SESSION_STORE: dict[str, dict] = {}

class InMemorySessionMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        # 1️⃣ 取得或產生 Session ID
        session_id = request.cookies.get("sid")
        if not session_id or session_id not in SESSION_STORE:
            session_id = str(uuid.uuid4())
            SESSION_STORE[session_id] = {"created": time.time(), "data": {}}
        # 2️⃣ 把 Session 資料掛到 request.state
        request.state.session = SESSION_STORE[session_id]["data"]
        # 3️⃣ 呼叫下游路由
        response: Response = await call_next(request)
        # 4️⃣ 回傳前寫回 Cookie（HttpOnly 提升安全）
        response.set_cookie(
            key="sid",
            value=session_id,
            httponly=True,
            max_age=3600,          # 1 小時過期
            samesite="lax",
        )
        return response

說明：此範例僅適合開發或測試環境，因為資料會隨程式重啟而遺失，且不具備分散式支援。

---

4.2 使用 Redis 作為 Session Store

# file: app/middleware/redis_session.py
import uuid
import json
from fastapi import Request, Response
from starlette.middleware.base import BaseHTTPMiddleware
import aioredis

REDIS_URL = "redis://localhost:6379/0"
SESSION_TTL = 1800  # 30 分鐘

class RedisSessionMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        redis = await aioredis.from_url(REDIS_URL, encoding="utf-8", decode_responses=True)

        # 1️⃣ 取得或建立 Session ID
        session_id = request.cookies.get("sid")
        if not session_id:
            session_id = str(uuid.uuid4())
            await redis.setex(session_id, SESSION_TTL, json.dumps({}))
        else:
            # 嘗試讀取現有 Session，若不存在則重新產生
            raw = await redis.get(session_id)
            if raw is None:
                session_id = str(uuid.uuid4())
                await redis.setex(session_id, SESSION_TTL, json.dumps({}))

        # 2️⃣ 解析 JSON 成 dict，掛到 request.state
        raw = await redis.get(session_id)
        request.state.session = json.loads(raw)

        # 3️⃣ 呼叫下游路由
        response: Response = await call_next(request)

        # 4️⃣ 把變更寫回 Redis，並更新過期時間
        await redis.setex(session_id, SESSION_TTL, json.dumps(request.state.session))

        # 5️⃣ 設定 Cookie（Secure 只在 HTTPS 時送出）
        response.set_cookie(
            key="sid",
            value=session_id,
            httponly=True,
            max_age=SESSION_TTL,
            secure=True,
            samesite="lax",
        )
        await redis.close()
        return response

要點：

• 使用 setex 同時寫入資料與 TTL，避免「孤兒 Session」。
• json.dumps/loads 是最簡單的序列化方式，若資料結構較複雜可改用 pickle（注意安全性）或 msgpack。

---

4.3 使用 PostgreSQL（SQLAlchemy）作為 Session Store

# file: app/models.py
from sqlalchemy import Column, String, DateTime, JSON, func
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class Session(Base):
    __tablename__ = "sessions"
    sid = Column(String, primary_key=True, index=True)           # Session ID
    data = Column(JSON, default=dict)                            # 任意 JSON 資料
    expires_at = Column(DateTime, nullable=False)               # 到期時間
    created_at = Column(DateTime, server_default=func.now())

# file: app/middleware/db_session.py
import uuid
from datetime import datetime, timedelta
from fastapi import Request, Response
from starlette.middleware.base import BaseHTTPMiddleware
from sqlalchemy.orm import Session as DBSession
from .models import Session as SessionModel
from .database import get_db  # 依賴注入的 DB Session

SESSION_TTL = timedelta(hours=2)

class DBSessionMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        # 取得依賴注入的 DB Session（同步範例）
        db: DBSession = next(get_db())

        # 1️⃣ 讀取或建立 Session
        sid = request.cookies.get("sid")
        now = datetime.utcnow()
        if sid:
            sess = db.query(SessionModel).filter(
                SessionModel.sid == sid,
                SessionModel.expires_at > now,
            ).first()
        else:
            sess = None

        if not sess:
            sid = str(uuid.uuid4())
            sess = SessionModel(
                sid=sid,
                data={},
                expires_at=now + SESSION_TTL,
            )
            db.add(sess)
            db.commit()
            db.refresh(sess)

        # 2️⃣ 把資料掛到 request.state
        request.state.session = sess.data

        # 3️⃣ 呼叫下游
        response: Response = await call_next(request)

        # 4️⃣ 更新 DB 中的 Session 資料與過期時間
        sess.data = request.state.session
        sess.expires_at = datetime.utcnow() + SESSION_TTL
        db.commit()

        # 5️⃣ 設定 Cookie
        response.set_cookie(
            key="sid",
            value=sid,
            httponly=True,
            max_age=int(SESSION_TTL.total_seconds()),
            secure=True,
            samesite="lax",
        )
        return response

說明：

• 這裡使用 SQLAlchemy 的 JSON 欄位直接保存 Python dict，適合 PostgreSQL、MySQL 8+。
• expires_at 欄位讓 DB 能自動過期（可搭配定期清理任務），避免無限累積。

---

4.4 在路由中使用 Session

# file: app/main.py
from fastapi import FastAPI, Request, Depends
from .middleware.redis_session import RedisSessionMiddleware
# 若使用 DB，改成 DBSessionMiddleware

app = FastAPI()
app.add_middleware(RedisSessionMiddleware)   # 或 InMemorySessionMiddleware / DBSessionMiddleware

@app.get("/login")
async def login(request: Request, username: str):
    # 假設已驗證使用者
    request.state.session["user_id"] = 123
    request.state.session["username"] = username
    return {"msg": f"歡迎 {username} 登入！"}

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

只要在任何路由 request.state.session 中讀寫，都會自動同步回後端儲存。

---

常見陷阱與最佳實踐

陷阱	可能的後果	建議的最佳實踐
Session ID 直接使用 UUID，但未加密或簽名	攻擊者可自行生成合法的 ID，造成 Session Fixation	使用 HMAC 簽名或 JWS（JWT）作為 ID，或在 Redis 中設定 key 前綴 防止猜測。
Cookie 未設 HttpOnly / Secure	JavaScript 可讀取，易受 XSS 攻擊	必加 httponly=True，在 HTTPS 環境下 secure=True。
Session TTL 設定過長	佔用過多儲存資源，且使用者登出後仍能使用舊 Session	依業務需求設定 合理的過期時間，並在登出時 主動刪除。
在高併發環境下忘記使用連線池	Redis/DB 連線耗盡，導致服務崩潰	使用 aioredis.from_url(..., pool_size=10) 或 SQLAlchemy engine = create_engine(..., pool_size=20)。
把大量資料放入 Session	讀寫速度下降，Cookie 大小限制無關但會讓 Redis/DB 壓力倍增	只存必要的鍵值（例如 user_id、role），其餘資料請放在資料庫或快取層。

---

實際應用場景

1. 電商網站的購物車

   • 使用者未登入時，將商品資訊寫入 Redis Session。
   • 登入後把 Session 中的 cart 合併至資料庫的永久購物車表。

2. 企業內部系統的 SSO

   • SSO 服務產生一次性 sid，透過 Redis 做短暫授權，過期時間設定為 10 分鐘，減少重複驗證的成本。

3. API 限流（Rate Limiting）

   • 以 sid 為鍵，統計單位時間內的請求次數，超過上限即回傳 429。Redis 的原子指令（INCR、EXPIRE）非常適合此需求。

4. 多服務（Microservice）間的共享會話

   • 所有服務共用同一個 Redis 集群，透過相同的 Cookie (sid) 即可在不同子域名間共享登入狀態。

---

總結

• Session 與 Cookie 分離 能讓我們在保持使用者體驗的同時，提升安全性與可擴展性。
• Redis 提供高速、過期自動管理，適合大流量、短期會話；資料庫 則適合需要永久保存或複雜查詢的情境。
• 透過 Middleware 把 Session 讀寫抽象化後，路由程式碼變得乾淨，只要操作 request.state.session 即可。
• 記得 設定 HttpOnly、Secure、SameSite，以及 合理的 TTL，才能避免常見的安全漏洞與資源浪費。

掌握以上概念與範例，你就能在 FastAPI 專案中快速導入可靠的 Session 後端，為後續的認證、授權、限流等功能奠定堅實基礎。祝開發順利！