Flask-RESTful——限流

一、Flask-RESTful如何限流？

如果是普通的Flask項目，怎麼使用節流，官方已經寫的很詳細了官網詳解

二、FBV——限流

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

app = Flask(__name__)
limiter = Limiter(
    app,
    key_func=get_remote_address,
    default_limits=["200 per day", "50 per hour"]
)
@app.route("/slow")
@limiter.limit("1 per day")
def slow():
    return ":("

@app.route("/medium")
@limiter.limit("1/second", override_defaults=False)
def medium():
    return ":|"

@app.route("/fast")
def fast():
    return ":)"

@app.route("/ping")
@limiter.exempt
def ping():
    return "PONG"
           

上面的Flask應用将具有以下速率限制特征：

通過請求的remote_address進行速率限制

預設速率限制為每天200，每小時50，适用于所有路線。

slow具有明确的速率限制修飾符的路由将繞過預設速率限制，并且每天僅允許1個請求。

該medium路由繼承預設限制，并增加了每秒1個請求的修飾限制。

該ping路由将不受任何預設速率限制的限制。

二、CBV——限流

from flask import Flask
from flask_limiter import Limiter
from flask_limiter.util import get_ipaddr
app = Flask(__name__)
limiter = Limiter(app,key_func=get_ipaddr)
           

from init import  limite
from flask_restful import Resource, reqparse, request
from tools.db_seesion import fetch_to_dict
from tools.Response import response
class User(Resource):
    paser = reqparse.RequestParser()
    paser.add_argument('password', type=min_max_length_str(5, 10), required=True, help='{error_msg}')
    paser.add_argument('email', required=True, help='email is required')
    '''
        如果是CVB 模式，直接加裝飾器limiter會無效，需要以這種方式才能生效,decorators 是個清單，如果需要多重限制
    '''
    decorators = [limiter.limit('5/minute', error_message='visit too fast :5/minute'),
                  limiter.limit('10/day', error_message='visit too fast :10/day')]

    @verify_token
    def get(self, username):
        user = fetch_to_dict('select id,username,email from users where username=:username', {'username': username},
                             fecth='one')
        if user:
            auth_user_id = str(request.headers.get('auth')).split(',')[1]
            if str(user.get('id')) != auth_user_id:
                return response(code=1002, msg='no permission,you can request youself only')
            return response(data=user)
        else:
            return response(code=1001, msg='user is not exits')
           

IP方式節流

# IP擷取方式
from flask_limiter.util import get_ipaddr,get_remote_address
           

其它方式節流

def get_token():
    return str(request.headers.get('auth')).split(',')[0]

from tools.authenticate import get_token

limiter = Limiter(
    app,
    key_func=get_token,
)