FastAPI速率限制实现教程

简介

速率限制是Web开发中的一种重要机制，用于限制客户端对API的访问频率，防止滥用、保护服务器资源，并确保服务的稳定性和公平性。在FastAPI中，虽然没有内置速率限制功能，但可以通过第三方库轻松实现。本教程将指导你如何在FastAPI应用中实现速率限制，特别适合新手学习。

什么是速率限制？

速率限制是指在给定时间窗口内，对客户端（如用户或IP地址）访问API的次数进行限制。例如，你可以设置每分钟最多10次请求，以避免API被恶意攻击或资源耗尽。

为什么在FastAPI中需要速率限制？

• 安全：防止暴力攻击，如密码猜测或API滥用。
• 性能：限制请求频率，避免服务器过载，确保稳定运行。
• 公平性：为所有用户提供平等的访问机会，避免单用户独占资源。

安装依赖

我们将使用fastapi-limiter库，这是一个专门为FastAPI设计的速率限制扩展。首先，确保你已经安装了FastAPI和uvicorn（或其他ASGI服务器）。然后，通过pip安装fastapi-limiter：

pip install fastapi-limiter

如果提示需要依赖，可以安装更多库，如redis用于存储限制状态（fastapi-limiter默认使用内存存储，也支持Redis）。为了使用Redis，可以安装：

pip install redis

使用fastapi-limiter库实现速率限制

步骤1：导入和初始化FastAPI应用

首先，创建一个FastAPI应用，并导入fastapi-limiter。

from fastapi import FastAPI, HTTPException, Request
from fastapi_limiter import FastAPILimiter
from fastapi_limiter.depends import RateLimiter
import asyncio

# 初始化FastAPI应用
app = FastAPI()

# 初始化速率限制器（使用内存存储，适用于开发环境）
@app.on_event("startup")
async def startup():
    await FastAPILimiter.init(app)  # 使用默认配置

步骤2：定义速率限制规则并应用到路由

速率限制规则使用RateLimiter依赖，你可以指定时间窗口和请求次数。

from fastapi import Depends

# 定义一个速率限制规则：每分钟最多5次请求
@app.get("/limited-endpoint")
async def limited_endpoint(request: Request, rate_limiter: RateLimiter = Depends(RateLimiter(times=5, seconds=60))):
    """
    这是一个受速率限制的端点，每分钟最多允许5次访问。
    """
    return {"message": "访问成功！"}

步骤3：完整示例代码

下面是一个完整的FastAPI应用示例，展示了速率限制的实现。

from fastapi import FastAPI, Request, Depends
from fastapi_limiter import FastAPILimiter
from fastapi_limiter.depends import RateLimiter
import asyncio

app = FastAPI()

# 启动时初始化速率限制器
@app.on_event("startup")
async def startup():
    # 使用内存存储（默认），适用于简单场景
    await FastAPILimiter.init(app)
    # 如果使用Redis，可以这样配置：
    # await FastAPILimiter.init(app, storage_uri="redis://localhost:6379")

# 首页，无速率限制
@app.get("/")
async def home():
    return {"message": "欢迎访问FastAPI应用！"}

# 受速率限制的端点：每分钟最多3次请求
@app.get("/api/limited")
async def limited_api(request: Request, rate_limiter: RateLimiter = Depends(RateLimiter(times=3, seconds=60))):
    return {"message": "你已经访问了这个端点，请注意速率限制！"}

# 另一个受限制的端点：每小时最多10次请求
@app.get("/api/strict-limited")
async def strict_limited_api(request: Request, rate_limiter: RateLimiter = Depends(RateLimiter(times=10, seconds=3600))):
    return {"message": "严格限制的端点，每小时最多10次请求。"}

步骤4：运行应用

保存代码为main.py，并使用uvicorn运行：

uvicorn main:app --reload

然后访问 http://localhost:8000/ 查看主页，访问 http://localhost:8000/api/limited 测试速率限制。如果你在短时间内多次访问受限端点，FastAPI将返回 429 Too Many Requests 错误。

高级配置和最佳实践

使用Redis进行持久化存储

在生产环境中，建议使用Redis作为存储后端，以避免内存限制和重启丢失数据。配置如下：

@app.on_event("startup")
async def startup():
    # 连接到本地Redis实例
    await FastAPILimiter.init(app, storage_uri="redis://localhost:6379")

自定义速率限制策略

你可以根据IP地址、用户ID或其他标识符来应用不同的限制。例如，基于IP的限制：

from fastapi_limiter.depends import RateLimiter

@app.get("/custom-limited")
async def custom_limited(request: Request):
    # 从请求中提取IP地址作为标识符
    client_ip = request.client.host
    # 使用自定义规则：每分钟最多5次请求，针对每个IP
    await RateLimiter(times=5, seconds=60, identifier=client_ip).check(request)
    return {"message": "基于IP的速率限制生效了！"}

错误处理和响应自定义

当速率限制触发时，FastAPI会自动返回HTTP 429状态码。你可以自定义错误响应：

from fastapi.responses import JSONResponse

@app.exception_handler(429)
async def rate_limit_exceeded(request: Request, exc):
    return JSONResponse(
        status_code=429,
        content={"detail": "请求过于频繁，请稍后再试！"}
    )

总结

在FastAPI中实现速率限制是保护API安全和性能的关键步骤。通过使用fastapi-limiter库，你可以轻松地设置和配置限制规则。本教程覆盖了基本安装、路由应用、高级配置和错误处理，适合新手快速上手。建议根据实际需求调整限制策略，并在生产环境中使用Redis存储以提高可靠性。

通过实施速率限制，你可以显著提升应用的稳定性和用户体验。如有更多问题，参考FastAPI和fastapi-limiter的官方文档。