FastAPI 结合 Redis 实现用户登录 Token 校验详解

在现代 Web 应用中，用户登录后的身份校验通常采用 Token（如 JWT）机制。为了提升性能和安全，很多场景会将 Token 存储在 Redis 这类高性能缓存中。本文将详细介绍如何用 FastAPI 结合 Redis 实现用户登录的 Token 校验功能，包括原理、流程、代码实现、最佳实践和安全建议。

一、方案原理与流程

用户登录：用户提交用户名和密码，后端校验通过后生成 Token（如 JWT 或自定义字符串）。
Token 存储：将 Token 及其关联的用户信息、过期时间等存入 Redis。
请求校验：用户后续请求需携带 Token，后端从 Redis 校验 Token 是否有效。
Token 续期/注销：支持 Token 自动续期或主动注销，提升安全性。

二、环境准备

Python 3.7+
FastAPI
aioredis（异步 Redis 客户端）
uvicorn

安装依赖：

pip install fastapi uvicorn aioredis[uvloop]

三、代码实现

1. Redis 连接管理

推荐使用 aioredis 实现异步 Redis 连接池。

import aioredis
from fastapi import FastAPI

app = FastAPI()

@app.on_event("startup")
async def startup_event():
    app.state.redis = await aioredis.from_url(
        "redis://localhost:6379", decode_responses=True
    )

@app.on_event("shutdown")
async def shutdown_event():
    await app.state.redis.close()

2. 用户登录与 Token 生成

这里以简单的用户名密码校验和 UUID 生成 Token 为例，实际项目可用 JWT。

import uuid
from fastapi import HTTPException, status
from pydantic import BaseModel

class LoginRequest(BaseModel):
    username: str
    password: str

@app.post("/login")
async def login(data: LoginRequest):
    # 假设用户名密码校验通过
    if data.username == "admin" and data.password == "123456":
        token = str(uuid.uuid4())
        # 存入 Redis，设置过期时间（如1小时）
        await app.state.redis.set(f"token:{token}", data.username, ex=3600)
        return {"token": token}
    else:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="用户名或密码错误")

3. Token 校验依赖

通过 FastAPI 的依赖注入机制实现 Token 校验。

from fastapi import Depends, Request

def get_token_from_header(request: Request):
    auth = request.headers.get("Authorization")
    if not auth or not auth.startswith("Bearer "):
        raise HTTPException(status_code=401, detail="缺少或非法的Token")
    return auth.split(" ", 1)[1]

async def verify_token(token: str = Depends(get_token_from_header), request: Request = None):
    username = await request.app.state.redis.get(f"token:{token}")
    if not username:
        raise HTTPException(status_code=401, detail="Token无效或已过期")
    return username

4. 受保护接口示例

@app.get("/profile")
async def profile(username: str = Depends(verify_token)):
    return {"user": username, "msg": "登录校验通过，获取到用户信息"}

5. Token 注销（登出）

@app.post("/logout")
async def logout(token: str = Depends(get_token_from_header), request: Request = None):
    await request.app.state.redis.delete(f"token:{token}")
    return {"msg": "已登出"}

四、最佳实践与安全建议

Token 设计：生产环境建议使用 JWT，包含签名和过期时间，防篡改。
HTTPS：所有接口务必启用 HTTPS，防止 Token 被窃取。
Token 续期：可在用户活跃时自动续期，提升体验。
多端登录控制：可用 Redis 的 Set 结构支持多端 Token 管理。
黑名单机制：支持 Token 拉黑，防止已注销 Token 被滥用。
异常处理：统一处理 Token 过期、无效等异常，提升安全性和用户体验。

五、总结

结合 FastAPI 和 Redis，可以高效实现用户登录的 Token 校验功能，兼顾性能与安全。通过依赖注入、异步编程和合理的 Token 管理机制，能大幅提升后端服务的健壮性和扩展性。

如需进一步扩展，可集成 JWT、OAuth2、单点登录等更复杂的认证方案。

Yiyuan