FastAPI 实现用户认证鉴权的方案，核心目标是在接口中统一获取并传递当前登录用户信息，避免重复编写认证逻辑，适配不同开发习惯。
Depends 依赖注入
核心方式：通过 FastAPI 内置的Depends()实现依赖注入，定义通用的get_current_user依赖函数，封装用户认证逻辑（模拟从 token 解析用户）； 使用方式：路由函数通过参数声明current_user: dict = Depends(get_current_user)，FastAPI 会自动执行依赖函数，将返回的用户信息注入到参数中； 核心优势：官方推荐、天然支持同步 / 异步、可多层依赖嵌套、参数自动解析校验，多个路由可直接复用同一依赖，代码简洁规范。
from fastapi import FastAPI, Depends
app = FastAPI()
# 定义可复用的依赖
def get_current_user(token: str = Depends(get_token)):
# 模拟用户认证逻辑
return {"user_id": 1, "username": "test_user"}
# 路由1复用依赖
@app.get("/user/info")
def get_user_info(current_user: dict = Depends(get_current_user)):
return current_user
# 路由2复用同一个依赖
@app.get("/user/orders")
def get_user_orders(current_user: dict = Depends(get_current_user)):
return {"user_id": current_user["user_id"], "orders": []}
装饰器实现
核心方式：通过functools.wraps自定义require_user装饰器，在装饰器内部封装认证逻辑（模拟从 Header 头解析 token、获取用户）； 使用方式：路由函数上方添加@require_user装饰器，装饰器会将解析后的用户信息作为current_user参数传入路由函数； 核心特点：Python 原生语法、自定义程度高，适合习惯装饰器写法的场景，但需手动处理参数传递、请求头解析，无 FastAPI 原生的依赖嵌套能力。
from fastapi import FastAPI, Header
from functools import wraps
app = FastAPI()
# 定义装饰器版依赖
def require_user(func):
@wraps(func)
async def wrapper(*args, **kwargs):
# 模拟获取token和用户
authorization = kwargs.get("authorization") or Header(...)
token = authorization.split(" ")[1]
current_user = {"user_id": 1, "username": "test_user"}
# 把用户信息传入函数
return await func(*args, current_user=current_user, **kwargs)
return wrapper
# 使用装饰器
@app.get("/user/info")
@require_user
def get_user_info(current_user: dict):
return current_user
用Proxy模式在Middleware中自动处理session
基于 FastAPI + Middleware + Session + Redis 实现 Session 用户数据存储。并实现轻量级鉴权即「登录生成 Session、接口鉴权、登出销毁 Session」。
Python在访问或设置 Proxy 计算属性时，可以触发一段预定义的计算逻辑。实现Session对象的脏检查与自动更新到Redis。这里我们会将创建的session对象挂载在request.state上：
request.state 是 FastAPI（基于 Starlette 框架）为单次 HTTP 请求提供的专属临时状态容器，核心作用是在单次请求的生命周期内，为不同处理环节（依赖、中间件、路由函数、自定义方法）提供一个统一的「数据共享空间」，用于存储和传递该请求的专属临时数据。简单来说：request.state 就是单个 HTTP 请求的「全局临时变量」，仅服务于当前请求，请求结束后会被框架自动销毁，完全隔离其他请求的数椐。
分布式服务
往session中存数据，不需要手动触发保存即可自动更新到redis。这样FastAPI服务就是无状态的，即使切换到其它Web服务，也能取到用户数据。比如：
# 保存用户数据
@router.post("/save")
async def save(data: ReservationModel, request: Request):
exception = await save_reservation(data)
request.state.session.reservation = data
return { "error": exception }
# 获取用户session中的数据
@router.get("/current")
async def current(request: Request):
return { "result": request.state.session.reservation or {}}
Session Proxy实现
这段代码实现了一个带「脏检测（Dirty Check）」的 Session 代理类，核心作用是封装原始字典类型的 Session 数据，自动检测 Session 内容是否被修改，并通过_dirty标记记录修改状态（用于后续按需持久化 Session，如仅当 Session 修改时才同步到 Redis/MongoDB，避免无意义的 IO 操作）。
class SessionProxy:
def __init__(self, session_id: str, session: dict[str, any], dirty = False):
# 用 super() 避免触发自身 __setattr__ 递归
super().__setattr__("_session_id", session_id)
super().__setattr__("_session", session)
super().__setattr__("_dirty", dirty)
def __setattr__(self, name, value):
super().__setattr__("_dirty", True)
self._session[name] = value
def __getattr__(self, name):
session = super().__getattribute__("_session")
return self._session[name] if name in session else None
基于 Redis 的 Session 管理中间件
这段代码实现了FastAPI/Starlette 专属的 Redis 持久化 Session 管理中间件，核心基于 BaseHTTPMiddleware 实现全局请求拦截，整合了SessionID 生成 / 解析、Redis 数据读写、SessionProxy 脏检测、Cookie 管理、数据序列化 / 反序列化等全套 Session 核心能力，最终实现用户会话的跨请求持久化（登录态保持、用户数据跨接口共享）。
is_session_required: 过滤无需校验 Session 的接口
import json
import uuid
from fastapi import Request, Response
from fastapi.responses import FileResponse
from starlette.middleware.base import BaseHTTPMiddleware
from redis import asyncio as aioredis
from middleware.session_proxy import SessionProxy
from common.custom_json_encoder import CustomJSONEncoder
# cd hotel\package\core>
# corepy\Scripts\activate
# pip install redis
# pip freeze > requirements.txt
class SessionRedisManager(BaseHTTPMiddleware):
def __init__(self, app, dispatch = None):
super().__init__(app, dispatch)
# Redis client
self.redis_client = aioredis.Redis(host='localhost', port=6379)
# Cookie中Session ID的名称
self.session_cookie_name = "HOTEL_SID"
# Session过期时间（1小时）
self.session_expire_seconds = 3600
# Redis中Session的key前缀
self.session_redis_prefix = "HOTEL_SESSION:"
def is_session_required(self, url_path: str):
if url_path.startswith("/api/v1/user/sign"):
return False
return True
def create_session_id(self):
return str(uuid.uuid4())
def get_session_id(self, request: Request):
return request.cookies.get(self.session_cookie_name)
# JWT token 存储在header中
# return request.headers.get(self.session_cookie_name)
async def update_session(self, request: Request):
session: SessionProxy = request.state.session
if not session._session_id or not isinstance(session._session, dict):
raise Exception("Seesion data error")
session_key = f"{self.session_redis_prefix}{session._session_id}"
if session._dirty:
# 对session._session中所有值做JSON序列化，处理JSON/非基础类型值
serialized_session = {
k: json.dumps(v, ensure_ascii=False, cls=CustomJSONEncoder)
for k, v in session._session.items()
}
print("update_session", session._session, serialized_session)
await self.redis_client.hmset(session_key, serialized_session)
await self.redis_client.expire(session_key, self.session_expire_seconds)
async def get_session_data(self, request) -> tuple[str|None, dict|None]:
session_id = self.get_session_id(request)
if session_id is None or len(session_id) < 32:
return (None, None)
raw_session = await self.redis_client.hgetall(f"{self.session_redis_prefix}{session_id}")
# 统一反序列化为原始值
session_obj = {
# Redis 哈希操作返回的键值默认均为 bytes 类型，
k.decode(): json.loads(v)
for k, v in raw_session.items()
}
return (session_id, session_obj)
async def dispatch(self, request, call_next):
if not request.url.path.startswith("/api/"):
return await call_next(request)
# 获取 session 数据
(session_id, session_data) = await self.get_session_data(request)
if not session_id:
# 如果是登录接口 session 不存在
if self.is_session_required(request.url.path):
session_id = self.create_session_id()
session_data = {}
# 初始化新 session 数据
request.state.session = SessionProxy(session_id, session_data)
response = await call_next(request)
# 不设置max_age/expires，Cookie仅在当前浏览器Session有效
response.set_cookie(
key = self.session_cookie_name,
value = session_id,
httponly = True, # 禁止客户端 cookie 访问
# secure = True,
samesite = "lax", # 防止CSRF攻击
)
# update session
await self.update_session(request)
return response
# 如果是非登录接口 session id 不存在，非法请求
else:
return Response("Bad Request", 400)
# 如果Session数据不存在，会话过期
if self.is_session_required(request.url.path) and not session_data:
return Response("Unauthorized", 401)
# 初始化新 session 数据
request.state.session = SessionProxy(session_id, session_data or {})
response = await call_next(request)
await self.update_session(request)
return response
使用时将此中间件加载到app上即可
app.add_middleware(SessionRedisManager)
