flask_limiter 模块实现 API 限流

代码

下面展示如何在 Flask 应用中使用flask_limiter 模块实现 API 限流

from flask import Flask, jsonify
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address

app = Flask(__name__)

# 初始化限流器，使用客户端 IP 作为标识
limiter = Limiter(
    app=app,
    key_func=get_remote_address,  # 默认使用客户端 IP
    default_limits=["100 per hour", "10 per minute"]  # 全局默认限制
)

# 路由1：普通限流（使用全局默认限制）
@app.route('/api/public')
@limiter.limit("5 per minute")  # 覆盖全局设置，更严格的限制
def public_api():
    return jsonify({"message": "Public API"})

# 路由2：根据条件动态限流
@app.route('/api/premium')
def premium_api():
    # 动态设置不同用户组的限制
    if is_premium_user():
        limiter.limit("30 per minute")  # 高级用户限制
    else:
        limiter.limit("3 per minute")   # 普通用户限制
    return jsonify({"message": "Premium API"})

def is_premium_user():
    """模拟检查用户权限的逻辑"""
    # 实际应用中这里会有用户验证逻辑
    return True  # 示例中直接返回 True

# 路由3：豁免限流的白名单
@app.route('/api/internal')
@limiter.exempt  # 完全免除限流
def internal_api():
    return jsonify({"message": " unlimit API "})

# 自定义错误处理器
@app.errorhandler(429)
def ratelimit_handler(e):
    return jsonify({
        "error": "over_limit",
        "message": f"over limit rate: {e.description}",
        "retry_after": e.retry_after
    }), 429

if __name__ == '__main__':
    app.run(debug=True)

关键功能说明：

1. 基本配置

   limiter = Limiter(
       key_func=get_remote_address,
       default_limits=["100 per hour", "10 per minute"]
   )
   

   • key_func: 识别客户端的依据（默认使用 IP）
   • default_limits: 全局默认限制（每小时100次 + 每分钟10次）

2. 路由级限流

   @limiter.limit("5 per minute")
   

   可用的限制格式：

   • "100/day"
   • "10 per hour"
   • "1/second"
   • "1000/hour;50/minute"（组合限制）

3. 动态限流

   if is_premium_user():
       limiter.limit("30 per minute")
   else:
       limiter.limit("3 per minute")
   

   根据用户状态动态设置不同限制

4. 豁免限流

   @limiter.exempt
   

5. 自定义错误响应

   @app.errorhandler(429)
   

安装依赖：

pip install flask flask-limiter

测试方法：

1. 使用浏览器快速刷新 /api/public 查看限流效果
2. 使用 curl 测试：

   # 连续访问测试
   for i in {1..6}; do curl http://localhost:5000/api/public; echo; done
   

代码：flask_limiter_test.py