SlowAPI Redis 限流后端

示例说明

限流的公平性取决于键的选择：纯 IP 会误伤共享出口，API key 才是真正的调用主体。key_func 按优先级取键——有 API key 用 key 维度，内网地址归入统一低配额类，其余回退 IP。装饰器的 cost 参数让重接口一次扣两点配额，比一刀切的次数限制更贴近真实成本。

技术要点

• key_func 分层取键保公平
• cost 参数按接口成本扣减配额
• headers_enabled 输出限流响应头

代码解读

limiter 全局默认每分钟三百次并启用 Redis 存储；search 接口六十次每分钟，report 每小时十次但每次扣两点。健康探针不加装饰器不消耗配额。RateLimitExceeded 的处理器把超限响应转成标准错误格式，响应头携带剩余配额与重置时间。

使用建议

限流键与鉴权体系对齐，未认证流量用更紧的默认。Redis 故障时确认降级行为符合预期。对突发接口可配突发容量而非纯固定窗口。

代码示例

from fastapi import FastAPI, Request
from slowapi import Limiter, _rate_limit_exceeded
from slowapi.errors import RateLimitExceeded
from slowapi.util import get_remote_address
from starlette.responses import JSONResponse

limiter = Limiter(
    key_func=get_remote_address,
    storage_uri='redis://localhost:6379/2',
    default_limits=['300/minute'],
    headers_enabled=True,
)

app = FastAPI()
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded)


def user_key(request: Request) -> str:
    api_key = request.headers.get('x-api-key')
    if api_key:
        return f'key:{api_key}'
    if request.client and request.client.host.startswith('10.'):
        return 'internal'  # 内部调用统一低配额
    return f'ip:{get_remote_address(request)}'


@app.get('/search')
@limiter.limit('60/minute', key_func=user_key)
async def search(request: Request, q: str = ''):
    return {'query': q, 'results': 12}


@app.get('/report')
@limiter.limit('10/hour', key_func=user_key, cost=2)
async def report(request: Request):
    return {'generated': True}


@app.get('/healthz')
async def healthz():
    return {'status': 'up'}  # 未加装饰器：探针不限流