技术文档

Appearance

65. FastAPI 结合 Redis 使用 ​

Redis 常被当作"放在内存里的公共小仓库"，用来做缓存、计数器、会话这类需要快读快写的数据。本篇用一个完整项目，从配置到封装再到三个真实场景，一步步把 Redis 接进 FastAPI。

一、Redis 基础配置 ​

安装准备 ​

text
uv add redis pydantic-settings
  • redis 是官方客户端，异步接口在 redis.asyncio（旧项目里的 aioredis 已合并进来，不要再单独装）。
  • 确保本地 Redis 已启动，可用 redis-cli ping 验证（返回 PONG 即正常）。没装 Redis 可用 Docker 快速起一个。

连接参数 ​

参数含义默认值
host主机地址127.0.0.1
port端口6379
password密码，无则留空空
db数据库编号0

db 是 Redis 内部的逻辑库编号（0~15），不是 MySQL 那种独立数据库，仅用于把不同业务的键隔离开。

集中管理配置 ​

用 pydantic-settings 把连接参数集中到一个 Settings，从 .env 读取。

core/config.py：

python
from functools import lru_cache

from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    model_config = SettingsConfigDict(
        env_file=".env", env_file_encoding="utf-8", extra="ignore"
    )

    redis_host: str = "127.0.0.1"
    redis_port: int = 6379
    redis_password: str = ""
    redis_db: int = 0


@lru_cache
def get_settings() -> Settings:
    return Settings()

.env.example（复制为 .env 后按需修改）：

text
REDIS_HOST=127.0.0.1
REDIS_PORT=6379
REDIS_PASSWORD=
REDIS_DB=0

get_settings() 用 @lru_cache 缓存实例，整个进程只解析一次 .env。密码这类敏感信息走环境变量，不硬编码进代码。

二、封装 RedisClient 模块 ​

把连接和常用操作封装成一个类，路由和依赖注入直接复用，业务代码不再关心底层的 redis.asyncio 细节。

core/redis_client.py：

python
import json

import redis.asyncio as aioredis

from core.config import get_settings


class RedisClient:
    def __init__(self) -> None:
        self._settings = get_settings()
        self._client: aioredis.Redis | None = None

    async def connect(self) -> None:
        self._client = aioredis.Redis(
            host=self._settings.redis_host,
            port=self._settings.redis_port,
            password=self._settings.redis_password or None,
            db=self._settings.redis_db,
            decode_responses=True,
        )
        await self._client.ping()

    async def close(self) -> None:
        if self._client is not None:
            await self._client.aclose()
            self._client = None

    @property
    def client(self) -> aioredis.Redis:
        if self._client is None:
            raise RuntimeError("Redis 尚未连接，请先调用 connect()")
        return self._client

    # ---------- 基础操作 ----------
    async def get(self, key: str) -> str | None:
        return await self.client.get(key)

    async def set(self, key: str, value: str, expire: int | None = None) -> bool:
        return await self.client.set(key, value, ex=expire)

    async def delete(self, *keys: str) -> int:
        return await self.client.delete(*keys)

    async def exists(self, key: str) -> bool:
        return bool(await self.client.exists(key))

    async def expire(self, key: str, seconds: int) -> bool:
        return await self.client.expire(key, seconds)

    async def incr(self, key: str) -> int:
        return await self.client.incr(key)

    # ---------- JSON 便捷方法 ----------
    async def set_json(self, key: str, value, expire: int | None = None) -> bool:
        return await self.client.set(
            key, json.dumps(value, ensure_ascii=False), ex=expire
        )

    async def get_json(self, key: str):
        data = await self.client.get(key)
        return json.loads(data) if data is not None else None


redis_client = RedisClient()


async def get_redis() -> RedisClient:
    return redis_client

要点：

  • decode_responses=True 让 get 直接返回字符串而不是 bytes。
  • get_json / set_json 内部用 json.dumps / json.loads 处理字典，因为标准 Redis 只存字符串。
  • 末尾的 redis_client 是单例，get_redis 是给 Depends 用的依赖函数。

main.py 在 lifespan 里连接、关闭：

python
from contextlib import asynccontextmanager

from fastapi import FastAPI

from core.redis_client import redis_client
from routers import redis_demo


@asynccontextmanager
async def lifespan(app: FastAPI):
    await redis_client.connect()
    yield
    await redis_client.close()


app = FastAPI(lifespan=lifespan)
app.include_router(redis_demo.router)

三、实际使用场景 ​

三个场景都写在 routers/redis_demo.py，通过 Depends(get_redis) 拿到同一个 RedisClient 实例。

python
import asyncio
from uuid import uuid4

from fastapi import APIRouter, Depends, HTTPException

from core.redis_client import RedisClient, get_redis

router = APIRouter(tags=["Redis 示例"])


# ---------- 场景1：缓存 ----------
async def load_from_db(user_id: int):
    await asyncio.sleep(1)  # 模拟一次慢查询
    return {"id": user_id, "name": f"用户{user_id}"}


@router.get("/cache/user/{user_id}")
async def get_user(user_id: int, redis: RedisClient = Depends(get_redis)):
    key = f"cache:user:{user_id}"
    cached = await redis.get_json(key)
    if cached is not None:
        return {"source": "cache", **cached}
    data = await load_from_db(user_id)
    await redis.set_json(key, data, expire=60)
    return {"source": "db", **data}


# ---------- 场景2：计数器 / 限流 ----------
@router.get("/counter/visit")
async def visit(redis: RedisClient = Depends(get_redis)):
    total = await redis.incr("counter:visit")
    return {"visit_count": total}


@router.get("/counter/limited")
async def limited(redis: RedisClient = Depends(get_redis)):
    key = "counter:limited"
    count = await redis.incr(key)
    if count == 1:
        await redis.expire(key, 60)  # 首次访问后 60 秒过期
    if count > 5:
        raise HTTPException(status_code=429, detail="请求过于频繁")
    return {"allowed": True, "count": count}


# ---------- 场景3：会话存储 ----------
@router.post("/session/login")
async def login(name: str, redis: RedisClient = Depends(get_redis)):
    token = uuid4().hex
    await redis.set_json(f"session:{token}", {"name": name}, expire=3600)
    return {"token": token}


@router.get("/session/me")
async def me(token: str, redis: RedisClient = Depends(get_redis)):
    data = await redis.get_json(f"session:{token}")
    if data is None:
        raise HTTPException(status_code=401, detail="会话不存在或已过期")
    return data

场景1：缓存 ​

第一次请求会走"慢查询"（约 1 秒），并把结果写进 Redis；之后 60 秒内直接命中缓存。

text
curl.exe http://127.0.0.1:8000/cache/user/1

第一次返回（约 1 秒，source 是 db）：

json
{ "source": "db", "id": 1, "name": "用户1" }

再请求一次（几乎瞬间，source 是 cache）：

json
{ "source": "cache", "id": 1, "name": "用户1" }

场景2：计数器 ​

incr 是原子的自增，天然适合计数和简单限流。

text
curl.exe http://127.0.0.1:8000/counter/visit

连续请求依次返回 {"visit_count": 1}、{"visit_count": 2}、{"visit_count": 3}……

限流接口连续请求 6 次：

text
curl.exe http://127.0.0.1:8000/counter/limited

前 5 次返回 {"allowed": true, "count": N}，第 6 次返回 429：

json
{ "detail": "请求过于频繁" }

场景3：会话存储 ​

登录生成一个带 1 小时过期时间的 token：

text
curl.exe "http://127.0.0.1:8000/session/login?name=张三"

返回：

json
{ "token": "6a2c...e9f1" }

带上 token 查询会话：

text
curl.exe "http://127.0.0.1:8000/session/me?token=6a2c...e9f1"

返回 {"name": "张三"}。token 不存在或过期后，返回 401。

四、注意事项 ​

连接管理 ​

  • 用 lifespan 创建、关闭连接，不要在每个请求里 connect() / 新建客户端。redis.asyncio.Redis 自带连接池，复用连接才有意义。
  • 多 worker（uvicorn --workers N）时，每个进程各自持有连接；不要把连接对象跨进程共享。
  • close() 只在应用退出时调用一次，不要放进请求路径。

异常处理 ​

缓存属于"锦上添花"，Redis 不可用时不应拖垮主流程。读缓存建议降级处理：

python
async def safe_get_json(redis, key: str):
    try:
        return await redis.get_json(key)
    except Exception:
        return None  # Redis 异常时降级，保证接口可用

写缓存失败也应吞掉异常或只记日志，让业务继续。是否降级取决于场景：会话、限流这类强依赖 Redis 的功能，失败时则要明确报错。

键命名规范 ​

  • 用冒号分层：业务:模块:标识，例如 cache:user:1、counter:visit、session:<token>。
  • 统一前缀便于按前缀清理、监控，也避免不同业务键名冲突。
  • 生产环境不要用 KEYS cache:* 全量扫描（会阻塞 Redis），用 SCAN 游标遍历。
  • 合理设置过期时间，避免缓存键无限增长占满内存。

其他 ​

  • 更新数据时优先"删除缓存"而不是"更新缓存"，让下一次读取自然重建，减少缓存与数据库不一致。
  • 密码走环境变量，不写进代码或提交到仓库。

练习：给缓存接口加一个 DELETE /cache/user/{user_id} 主动删除缓存；把限流阈值和窗口改成可配置；用 redis-cli 查看 cache:user:1 的剩余过期时间（TTL cache:user:1）。

网站备案号:津ICP备15001525号-2