pip install Flask-Limiter
Flask-Limiter官方文档
基本使用
默认是用IP作为key进行计数的,你也可以自定义key,具体看官网
from flask import Flask
from flask_limiter import Limiter
from flask_limiter.util import get_remote_addressapp = Flask(__name__)limiter = Limiter(get_remote_address,app=app,default_limits=["200 per day", "50 per hour"],storage_uri=f'redis://:{REDIS_PASSWORD}@{REDIS_HOST}:{REDIS_PORT}/{REDIS_DATABASE}'
)# 或者先在一个global的地方初始化:
limiter = Limiter(get_remote_address, storage_uri=REDIS_URL)
# 然后在 create_app 的地方 init_app
limiter.init_app(flask_app)# 视图函数里面加装饰器
@limiter.limit("20/day;5/minute")
def post(self):
自定义Key Functions
自带的get_remote_address不好用,无法获取真是用户的IP,需要自己写一个Key Functions
from flask import request# request.headers.X-Forwarded-For: 真.实.I.P, 192.168.2.2, 第二台代理机器的IP, ..., 最后一台代理机器的IP
def get_real_ip():if x_forwarded_for := request.headers.get("X-Forwarded-For"):if ips := list(filter(lambda x: x, [ip.strip() for ip in x_forwarded_for.split(',')])):return ips[0]if x_real_ip := request.headers.get("X-Real-Ip"):return x_real_ip.strip()if remote_addr := request.remote_addr:return remote_addr.strip()return 'ipunknown'# 全局 Key Functions
limiter = Limiter(get_real_ip, storage_uri=config.REDIS_RATE_LIMIT_URL)# 局部 Key Functions
@limiter.limit("1 per day", key_func = lambda : current_user.username)
def post(self):
自定义错误处理
def too_many_request(request_limit):return jsonify({'code': 403, 'msg': '请稍后再试!'})# 全局错误处理
limiter = Limiter(get_remote_address, storage_uri=REDIS_URL, on_breach=too_many_request)# 单独错误处理
@limiter.limit("20/day;5/minute", on_breach=too_many_request)
def post(self):
自定义计数逻辑
比如 有时候,希望登录失败的,才进行计数,登录失败超过20次就禁止访问,而登录成功的,不计数,随便请求 无限制
@limiter.limit("20/day;5/minute", deduct_when=lambda response: response.json.get('code') ==400)
def post(self):# 上面的例子是:返回的JSON中code==400 就做一次计数,否则就不计数(这次请求,不计入限制)
# 还可以有其他的计数逻辑,比如
deduct_when=lambda response: response.status_code != 200
更多操作见官网