一、flask使用flask-limiter存在版本问题
Flask==1.1.4
Flask-Bootstrap==3.3.7.1
Flask-Caching==1.9.0
Flask-Cors==3.0.10
Flask-Limiter==1.4
Flask-Migrate==2.5.3
Flask-RESTful==0.3.8
Flask-Script==2.0.6
Flask-SocketIO==5.0.1
Flask-Sockets==0.2.1
Flask-SQLAlchemy==2.4.4
Jinjia2==2.11.3
Werkzeug==1.0.1
flask1.x.x版本,使用flask-limiter=1.4.0版本的。
二、配置
ext/__init__.py
from flask_sqlalchemy import SQLAlchemy
from flask_restful import Api
from flask_cors import CORS
from flask_socketio import SocketIO
from threading import Lock
from flask_caching import Cache
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address #以ip地址作为流量控制条件
db = SQLAlchemy() #数据库拓展
api = Api() #api拓展
# cors = CORS(cors_allowed_origins='*')
cors = CORS()
socketio = SocketIO(cors_allowed_origins='*') #websocket拓展
all_lock = Lock()
cache = Cache() #缓存
limiter = Limiter(key_func=get_remote_addr,default_limits=['120/minute'])
#流量控制拓展,设置了一个全局的流量控制,所有的接口都生效
apps/__init__.py
from flask import Flask
from ext import limiter
app = Flask(__file__)
limiter.init_app(app) #拓展流量控制
三、简单使用
频率限制语法:
1、'100 per day'、'20 per hour'、'5 per minute'、'1 per second'
2、'100/day'、'20/hour'、'5/minute'、'1/second' #推荐这个,看起来简单
FBV使用
from ext import limiter
from flask import jsonify,request
from apps import app
@limiter.limit('5/minute') #一分钟5次
@app.route('/test',method=['POST','GET'])
def test():
return jsonify({'code':200,'msg':'测试'})
CBV使用
from ext import limiter
from flask import jsonify,request
class TestRateView(views.MethodView):
decorators = [limiter.limit("5/minute")]
def get(self):
return view_response(code=200,msg='测试西安舒')
四、高级使用
在很多时候,我们不仅仅需要对IP进行流量控制,可能需要对用户名,对请求携带的数据进行流量控制,这时候就需要自定义流量控制的条件了
from base.authen import decode_token_for_limiter
from flask import request
#根据用户的id作为限制
def get_user_id():
token = request.headers.get('token')
if not token:
token = request.cookies.get('token')
if token:
#解析token,拿到用户的id
user = decode_token_for_limiter(token)
else:
user = 'anonymity'
#返回什么,就对什么进行流量控制
return user
使用:
from ext import limiter
from flask import jsonify,request
from util.limiter_ import get_user_id
class TestRateView(views.MethodView):
decorators = [limiter.limit("5/minute",key_func=get_user_id)]
def get(self):
return view_response(code=200,msg='测试西安舒')
五、处理流量控制的状态码
在开发过程中,一般需要统一状态码,或者使用自己的一套规则。一般需要重新修改状态码的。
可以使用flask自带的异常处理装饰器来实现:文章来源:https://www.toymoban.com/news/detail-623314.html
在启动文件中:超过频率限制时,状态码 是429,可以捕获这个状态码文章来源地址https://www.toymoban.com/news/detail-623314.html
from flask import jsonify
from apps import app
@app.errorhandler(429)
def handle_rate_limit_exceeded(e):
return jsonify({"error": "请求频率超过限制",'code':400}), 200
到了这里,关于Flask框架-流量控制:flask-limiter的使用的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!