中文
devexamples

SlowAPI Rate Limit Redis Backend

Throttle FastAPI endpoints with SlowAPI using a Redis backend, per-user limits, custom headers, and exempted internal callers.

PythonFastapiResilience

Code

snippet
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'}  # 未加装饰器：探针不限流

Related Examples