如何为FastAPI的SlowAPI限流器设置域名豁免?
使用FastAPI SlowAPI实现特定域名的限流豁免
可以通过自定义key_func结合多规则限流来实现特定域名的豁免,核心思路是:
- 从请求头(Origin/Referer)识别请求来源域名
- 对目标域名的请求返回特殊标识,为该标识设置无限(或极高阈值)的限流规则
- 其他请求使用常规限流标识(如客户端IP)并应用正常限流规则
注意事项
request.client.host返回的是客户端IP地址,无法直接判断域名,必须通过请求头(Origin或Referer)获取来源域名:
Origin:跨域请求中会携带,准确标识请求的源域名Referer:记录请求的来源页面URL,可作为Origin的补充
完整实现代码
from fastapi import FastAPI, Request from slowapi import Limiter, _rate_limit_exceeded_handler from slowapi.util import get_remote_address from slowapi.errors import RateLimitExceeded from urllib.parse import urlparse def extract_domain(url: str) -> str: """解析URL,提取域名部分""" parsed_url = urlparse(url) return parsed_url.netloc def my_key_func(request: Request): """自定义限流Key生成函数,豁免my_domain.com的请求""" # 检查Origin头(优先用这个,跨域请求更可靠) origin = request.headers.get("Origin") if origin: domain = extract_domain(origin) # 可根据需求添加子域名,比如www.my_domain.com if domain in ["my_domain.com", "www.my_domain.com"]: return "exempted" # 检查Referer头(补充非跨域场景) referer = request.headers.get("Referer") if referer: domain = extract_domain(referer) if domain in ["my_domain.com", "www.my_domain.com"]: return "exempted" # 其他请求用客户端IP作为限流Key return get_remote_address(request) # 初始化限流器 limiter = Limiter(key_func=my_key_func) app = FastAPI() app.state.limiter = limiter app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler) # 带限流规则的路由:默认10次/分钟,豁免请求不限流 @app.get("/limited-route") @limiter.limit("10/minute; exempted=inf/minute") async def limited_function(): return {"success": 200}
代码说明
- 域名解析:用
urlparse解析请求头中的URL,精准提取域名,避免字符串包含导致的误判 - Key生成逻辑:
- 来自目标域名的请求返回
"exempted"标识 - 其他请求返回客户端IP,作为限流的区分Key
- 来自目标域名的请求返回
- 限流规则:
"10/minute; exempted=inf/minute"表示:- 默认请求(非豁免)限制为10次/分钟
- 标识为
"exempted"的请求不受限流限制(inf表示无限) - 若担心
inf不兼容,可替换为极高阈值如"1000000/minute"
内容的提问来源于stack exchange,提问作者gbiz123
相关产品推荐
相关产品推荐

