如何在多文件结构FastAPI项目中实现接口限流（每秒10次请求）

可以用slowapi库来实现，这是FastAPI生态里常用的限流工具，下面是具体步骤：

1. 安装依赖 #

执行命令安装所需包：

pip install slowapi limits

2. 封装限流器（避免循环导入） #

新建一个专门的配置文件，比如dependencies/rate_limit.py，用来初始化限流器：

from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded

# 配置限流规则：每秒最多10次请求，用客户端IP作为限流标识
limiter = Limiter(key_func=get_remote_address, default_limits=["10/second"])

3. 配置FastAPI应用（main.py） #

修改main.py，把限流器挂载到应用上，并注册限流异常的处理器：

from .router import router
from fastapi import FastAPI
from .dependencies.rate_limit import limiter, _rate_limit_exceeded_handler, RateLimitExceeded

def get_application() -> FastAPI:
    application = FastAPI()
    
    # 挂载限流器到应用状态
    application.state.limiter = limiter
    # 注册限流超出后的异常处理
    application.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
    
    application.include_router(router)
    return application

app = get_application()

4. 给路由接口添加限流 #

有两种方式可以给router.py里的接口加限流：

方式一：单个接口单独限流 #

在router.py中导入限流器，给指定接口加装饰器：

from fastapi import APIRouter
from .dependencies.rate_limit import limiter

router = APIRouter()

@router.get("/")
@limiter.limit("10/second")  # 这里可以单独指定规则，和默认不同也可以
async def some_func():
    # 你的业务代码
    return {"message": "Hello World"}

方式二：给整个Router的所有接口统一限流 #

如果要让某个Router下的所有接口都遵循同一限流规则，不用逐个加装饰器，直接在main.py导入路由时配置依赖：

from fastapi import Depends
from .router import router
from fastapi import FastAPI
from .dependencies.rate_limit import limiter, _rate_limit_exceeded_handler, RateLimitExceeded

def get_application() -> FastAPI:
    application = FastAPI()
    
    application.state.limiter = limiter
    application.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
    
    # 给整个router添加限流依赖
    application.include_router(router, dependencies=[Depends(limiter.limit("10/second"))])
    return application

app = get_application()

补充说明 #

• get_remote_address是用客户端IP作为限流的标识，如果你需要基于用户ID等其他维度限流，可以自定义key_func，比如从请求中获取用户ID返回即可。
• 如果是分布式部署的场景，建议用Redis作为限流存储（需要额外安装redis包），替换默认的内存存储，避免多个实例之间限流规则不统一。

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