基于FastAPI与Firebase实现用户会话状态存储方案问询

场景说明 #

当前基于FastAPI+Firebase实现身份认证，希望避免每次请求手动传递JWT Token，也不想重复调用auth.verify_id_token(jwt)获取用户信息，需要类似Flask的session机制来管理登录用户的会话状态。

原始代码参考 #

from fastapi import APIRouter, Depends, HTTPException, Request
from fastapi.responses import JSONResponse
import pyrebase
import firebase_admin
from pathlib import Path

import sys
BASE_DIR = Path(__file__).resolve().parent.parent
sys.path.append(str(BASE_DIR))

from firebase_admin import auth
from config.config_settings import get_settings,get_firebase_auth
from database.mongoHandler import MongoDBHandler
from models.user import UserLogin,UserRegister


# Create an instance of the FastAPI router
authRouter = APIRouter()
settings = get_settings()
firebase_config = get_firebase_auth()
firebase = pyrebase.initialize_app(firebase_config)
mongo = MongoDBHandler()

# Initialize Firebase Admin SDK with credentials
firebase_admin.initialize_app(firebase_admin.credentials.Certificate(settings.FIREBASE_ADMIN_CREDENTIALS))



# Define a route for user registration
@authRouter.post("/register")
async def register_user(data: UserRegister):
    try:
        # Create a new user in Firebase Authentication
        user = auth.create_user(
            email=data.email,
            password=data.password
        )
        mongo.insert_one_document({
            "user_fullname": data.user_fullname,
            "email" : data.email
        })
        return {"message": "User registered successfully", "uid": user.uid}
    except auth.EmailAlreadyExistsError:
        raise HTTPException(status_code=400, detail="Email already registered")
    except Exception as e:
        raise HTTPException(status_code=500, detail="Registration failed")


@authRouter.post("/login")
async def login(data: UserLogin):
    try:
        user = firebase.auth().sign_in_with_email_and_password(
            email = data.email,
            password = data.password,
        )
        jwt = user['idToken']
        return JSONResponse(content={'token': jwt}, status_code=200)
    except Exception as e:
        # Return the error message in the response
        return HTTPException(detail={'message': str(e)}, status_code=400)

优化方案 #

方案1：Redis缓存会话（适合分布式部署） #

通过Redis存储会话ID与用户信息的映射，登录时生成唯一会话ID，将JWT和用户信息存入Redis，同时返回HttpOnly Cookie存储会话ID。后续请求通过Cookie自动携带会话ID，依赖项从Redis读取缓存的用户信息，无需重复验证JWT。

实现步骤：

1. 安装依赖：

pip install redis python-multipart

2. 配置Redis连接（可加入config_settings.py）：

from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    # 其他原有配置...
    REDIS_HOST: str = "localhost"
    REDIS_PORT: int = 6379
    REDIS_DB: int = 0
    SESSION_EXPIRE_SECONDS: int = 3600  # 会话有效期1小时

settings = Settings()

3. 初始化Redis客户端：

import redis

redis_client = redis.Redis(
    host=settings.REDIS_HOST,
    port=settings.REDIS_PORT,
    db=settings.REDIS_DB,
    decode_responses=True
)

4. 修改登录接口，生成会话并缓存用户信息：

import uuid
from fastapi import Response

@authRouter.post("/login")
async def login(data: UserLogin, response: Response):
    try:
        user = firebase.auth().sign_in_with_email_and_password(
            email=data.email,
            password=data.password,
        )
        jwt = user['idToken']
        # 验证JWT获取用户信息
        decoded_user = auth.verify_id_token(jwt)
        # 生成唯一会话ID
        session_id = str(uuid.uuid4())
        # 将用户信息和JWT存入Redis，设置过期时间
        redis_client.hset(
            f"session:{session_id}",
            mapping={
                "uid": decoded_user["uid"],
                "email": decoded_user["email"],
                "id_token": jwt
            }
        )
        redis_client.expire(f"session:{session_id}", settings.SESSION_EXPIRE_SECONDS)
        # 设置HttpOnly Cookie，防止前端XSS窃取
        response.set_cookie(
            key="session_id",
            value=session_id,
            httponly=True,
            max_age=settings.SESSION_EXPIRE_SECONDS,
            secure=True,  # 生产环境开启，仅HTTPS传输
            samesite="lax"
        )
        return {"message": "Login successful"}
    except Exception as e:
        raise HTTPException(detail={'message': str(e)}, status_code=400)

5. 编写依赖项自动获取当前用户：

from fastapi import Depends, Request

async def get_current_user(request: Request):
    session_id = request.cookies.get("session_id")
    if not session_id:
        raise HTTPException(status_code=401, detail="Not authenticated")
    # 从Redis读取会话信息
    session_data = redis_client.hgetall(f"session:{session_id}")
    if not session_data:
        raise HTTPException(status_code=401, detail="Session expired or invalid")
    # 可选：定期刷新会话有效期
    redis_client.expire(f"session:{session_id}", settings.SESSION_EXPIRE_SECONDS)
    return {
        "uid": session_data["uid"],
        "email": session_data["email"],
        "id_token": session_data["id_token"]
    }

6. 受保护接口使用依赖项：

@authRouter.get("/profile")
async def get_profile(current_user: dict = Depends(get_current_user)):
    # 直接使用current_user，无需重复验证JWT
    user_info = mongo.find_one_document({"email": current_user["email"]})
    return {"user": user_info}

方案2：FastAPI SessionMiddleware（轻量单实例场景） #

使用FastAPI内置的SessionMiddleware，基于加密Cookie存储会话数据，适合单实例部署，无需额外依赖Redis。

实现步骤：

1. 安装依赖：

pip install itsdangerous

2. 在主FastAPI应用中配置SessionMiddleware：

from fastapi import FastAPI
from fastapi.middleware.session import SessionMiddleware

app = FastAPI()
# 设置加密密钥，生产环境请使用随机生成的强密钥
app.add_middleware(SessionMiddleware, secret_key="your-strong-secret-key", max_age=3600)

3. 修改登录接口，将用户信息存入session：

@authRouter.post("/login")
async def login(data: UserLogin, request: Request):
    try:
        user = firebase.auth().sign_in_with_email_and_password(
            email=data.email,
            password=data.password,
        )
        jwt = user['idToken']
        decoded_user = auth.verify_id_token(jwt)
        # 将用户信息存入session
        request.session["user"] = {
            "uid": decoded_user["uid"],
            "email": decoded_user["email"],
            "id_token": jwt
        }
        return {"message": "Login successful"}
    except Exception as e:
        raise HTTPException(detail={'message': str(e)}, status_code=400)

4. 编写依赖项获取当前用户：

async def get_current_user(request: Request):
    user = request.session.get("user")
    if not user:
        raise HTTPException(status_code=401, detail="Not authenticated")
    return user

5. 受保护接口使用依赖项：

@authRouter.get("/profile")
async def get_profile(current_user: dict = Depends(get_current_user)):
    user_info = mongo.find_one_document({"email": current_user["email"]})
    return {"user": user_info}

方案对比 #

• Redis方案：支持分布式部署，会话数据存储在Redis，多实例共享；适合生产环境集群场景。
• SessionMiddleware方案：配置简单，无需额外服务；但会话数据存在Cookie中（加密），仅适合单实例部署，会话大小受Cookie限制。

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