如何在FastAPI中实现请求缓存？附代码示例

根据你的代码场景，分接口类型给出针对性的缓存方案，同时修正原代码中的SQL注入风险问题：

一、GET 首页接口的缓存 #

这个接口返回HTML模板，适合从客户端HTTP缓存和服务器端缓存两方面优化：

1. 客户端HTTP缓存（推荐） #

通过设置响应头让浏览器缓存页面，减少重复请求：

from fastapi import Response

@app.get("/")
def home(request: Request, response: Response):
    # 设置公共缓存，10分钟（600秒）过期
    response.headers["Cache-Control"] = "public, max-age=600"
    # 可选：添加ETag实现协商缓存，内容未变化时返回304
    response.headers["ETag"] = "home-page-v1"
    return templates.TemplateResponse("index.html", {"request": request})

• Cache-Control：告诉浏览器缓存该响应，max-age指定缓存时长；
• ETag：当浏览器再次请求时会携带If-None-Match头，服务器对比ETag一致则返回304，节省带宽。

2. 服务器端内存缓存 #

如果页面内容动态生成但更新频率低，可用内存缓存减少模板渲染开销：

from cachetools import TTLCache

# 缓存配置：最多存100条，10分钟过期
page_cache = TTLCache(maxsize=100, ttl=600)

@app.get("/")
def home(request: Request):
    cache_key = "home_page"
    # 若页面含用户专属内容，需将用户标识加入cache_key（比如request.session.get("user_id")）
    if cache_key in page_cache:
        return page_cache[cache_key]
    
    response = templates.TemplateResponse("index.html", {"request": request})
    page_cache[cache_key] = response
    return response

二、POST /api/getData 接口的缓存处理 #

你的该接口是插入数据到数据库的非幂等操作，绝对不能缓存请求触发的数据库写入逻辑，但可缓存其响应结果（如果响应内容可预测）：

1. 修正原代码的SQL注入风险（必须做） #

原代码用字符串拼接SQL存在严重安全问题，改用参数化查询：

from datetime import datetime

@app.post("/api/getData")
async def getData(request: Request, databody = Body()):
    data = databody["data"]
    
    with connection.cursor() as cursor:
        # 参数化查询避免SQL注入
        cursor.execute(
            "INSERT INTO database (ip, useragent, datetime) VALUES (%s, %s, %s)",
            (request.headers["host"], request.headers["user-agent"], datetime.now())
        )
        connection.commit()  # 别忘了提交事务
    
    # 不要直接返回Request对象（无法序列化），返回需要的字段
    return {
        "req": {
            "host": request.headers["host"],
            "user_agent": request.headers["user-agent"]
        }
    }

2. 服务器端响应缓存 #

如果同一请求参数的响应内容固定，可用缓存避免重复生成响应：

from cachetools import TTLCache

response_cache = TTLCache(maxsize=1000, ttl=300)  # 5分钟过期

@app.post("/api/getData")
async def getData(request: Request, databody = Body()):
    data = databody["data"]
    # 生成唯一缓存key：结合请求参数和客户端标识
    cache_key = f"getData_{data}_{request.headers.get('host', '')}_{request.headers.get('user-agent', '')}"
    
    if cache_key in response_cache:
        return response_cache[cache_key]
    
    # 执行数据库插入（已修正SQL注入）
    with connection.cursor() as cursor:
        cursor.execute(
            "INSERT INTO database (ip, useragent, datetime) VALUES (%s, %s, %s)",
            (request.headers["host"], request.headers["user-agent"], datetime.now())
        )
        connection.commit()
    
    response_data = {
        "req": {
            "host": request.headers["host"],
            "user_agent": request.headers["user-agent"]
        }
    }
    response_cache[cache_key] = response_data
    return response_data

3. 前端配合缓存 #

前端可将响应存在localStorage，减少重复请求：

async function fetchData(data) {
    const cacheKey = `getData_${JSON.stringify(data)}`;
    const cached = localStorage.getItem(cacheKey);
    
    if (cached) {
        const { data: cachedData, expires } = JSON.parse(cached);
        if (Date.now() < expires) {
            renderData(cachedData);
            return;
        }
        localStorage.removeItem(cacheKey); // 清除过期缓存
    }
    
    const res = await fetch('/api/getData', {
        method: 'POST',
        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify({ data })
    });
    const responseData = await res.json();
    
    // 存入缓存，设置5分钟过期
    localStorage.setItem(cacheKey, JSON.stringify({
        data: responseData,
        expires: Date.now() + 5 * 60 * 1000
    }));
    
    renderData(responseData);
}

function renderData(data) {
    // 你的页面渲染逻辑
    console.log(data);
}

三、进阶：分布式缓存（多实例部署场景） #

如果应用是多实例部署，内存缓存无法共享，可改用Redis作为分布式缓存：

from fastapi_cache import FastAPICache
from fastapi_cache.backends.redis import RedisBackend
from fastapi_cache.decorator import cache
from redis import asyncio as aioredis

# 启动时初始化Redis
@app.on_event("startup")
async def startup():
    redis = aioredis.from_url("redis://localhost")
    FastAPICache.init(RedisBackend(redis), prefix="fastapi-cache")

# 给GET接口加缓存（10分钟过期）
@app.get("/")
@cache(expire=600)
def home(request: Request):
    return templates.TemplateResponse("index.html", {"request": request})

# 给POST接口加缓存（仅当操作幂等时使用，这里不建议，除非你的插入逻辑是幂等的）
@app.post("/api/getData")
@cache(expire=300)
async def getData(request: Request, databody = Body()):
    # ... 数据库操作和响应生成 ...
    return response_data

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