當前位置：首頁 > 编程资源 > 综合教程 >内容正文

综合教程

Flask-Limiter使用详细说明

發布時間：2023/12/13 综合教程 48 生活家

生活随笔收集整理的這篇文章主要介紹了 Flask-Limiter使用详细说明小編覺得挺不錯的,現在分享給大家,幫大家做個參考.

Flask-Limit詳細說明

在flask項目中我們需要對全部或者一部分接口進行限制，又不想造輪子，那怎么辦呢？

所以這就是flask-limit出現的原因，不過對于相對復雜的需求，還是自己造輪子吧！

安裝與簡單使用

安裝：pip install Flask-Limiter

快速開始：

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"

上訴頻率限制說明：

默認通過請求的remote_address進行限制。
默認限制為200次/天，50次/小時；適用于所有路線
slow路由的限制將繞過默認的速率限制，為1次/天
medium路由繼承默認限制，并增加了1次/秒的限制
ping路由不受任何默認速率限制的約束

注意：靜態路由不受速率限制

每次請求超出速率限制時，將不會調用view函數，而是會引發429http錯誤。

速率限制規則：

[count] [per|/] [n (optional)] [second|minute|hour|day|month|year]

可以使用自己選擇的分隔符將多個速率限制組合起來。

示例：

10 per hour
10/hour
10/hour;100/day;2000 per year
100/day, 500/7days

使用的詳細說明

看完上面的部分其實已經滿足大部分需求了，但是真實的情況下，可能還存在其他的定制服務，以下就是詳細說明。

初始化

初始化有兩種方式：

使用構造函數

from flask_limiter import Limiter
from flask_limiter.util import get_remote_address
....

limiter = Limiter(app, key_func=get_remote_address)

使用延遲應用初始化 init_app

limiter = Limiter(key_func=get_remote_address)
limiter.init_app(app)

實際開發中更有可能使用的是延遲初始化。

裝飾器

我們所使用的是已創建的Limiter示例的limit方法，可根據喜好和使用場景，有以下幾種使用方式：

單裝飾

@app.route("....")
@limiter.limit("100/day;10/hour;1/minute")
def my_route()
  ...

多裝飾

@app.route("....")
@limiter.limit("100/day")
@limiter.limit("10/hour")
@limiter.limit("1/minute")
def my_route():
  ...

新增自定義的功能

下方會有詳細介紹此裝飾器內的參數的說明

def my_key_func():
  ...

@app.route("...")
@limiter.limit("100/day", my_key_func)
def my_route():
  ...

限制域

即指定根據什么進行限制，對應的參數為key_func，flask_limiter.util提供了兩種方式：

flask_limiter.util.get_ipaddr(): 使用X-Forwarded-For標頭中的最后一個IP地址，否則回退到請求的remote_address（不建議使用）
flask_limiter.util.get_remote_address(): 使用請求的remote_address。

在真實開發中，大部分項目都配備了Nginx，所以如果直接使用get_remote_address的話獲取到的是Nginx服務器的地址，非常危險！！！

所以項目中很有可能都是自定義key_func！

搭載Nginx服務器的key_func示例：

def limit_key_func():
    return str(flask_request.headers.get("X-Forwarded-For", '127.0.0.1'))

不過以上設置的依據還是根據Nginx的配置決定的，有興趣的同學還可以了解一下X-Forwarded-For和X-Real-IP的區別。

X-Forwarded-For 一般是每一個非透明代理轉發請求時會將上游服務器的ip地址追加到X-Forwarded-For的后面，使用英文逗號分割；

X-Real-IP一般是最后一級代理將上游ip地址添加到該頭中；

X-Forwarded-For是多個ip地址，而X-Real-IP是一個；

如果只有一層代理，這兩個頭的值就是一樣的。

所以上方自定義的方法僅作參考。

動態加載限制字符串

常見的限制規則已在上文介紹過，這里介紹的在某些情況下，需要從代碼外部的源（數據庫，遠程api等）中檢索速率限制。

def rate_limit_from_config():
    return current_app.config.get("CUSTOM_LIMIT", "10/s")

@app.route("...")
@limiter.limit(rate_limit_from_config)
def my_route():
    ...

所裝飾的路由上的每個請求都會調用提供的可調用對象。對于昂貴的檢索，請考慮緩存響應。

豁免條件

個人覺得這可以從兩個方面來談，一是針對key，一是針對計次，以下我們分別進行介紹。

白名單：

方式一：參數為exempt_when，設置這個參數將不被頻率限制。

@app.route("/expensive")
@limiter.limit("100/day", exempt_when=lambda: current_user.is_admin)
def expensive_route():
  ...

方式二：請求過濾器Limiter.request_filter()方法（沒研究）

@limiter.request_filter
def header_whitelist():
    return request.headers.get("X-Internal", "") == "true"

@limiter.request_filter
def ip_whitelist():
    return request.remote_addr == "127.0.0.1"

不計次情況：參數為deduct_when，判斷某些情況不計入使用頻率的次數。

def func_deduct(response):
    """
    頻率限制之根據response決定是否計次
    :param response: flask.wrappers.Response對象
    :return: 計次返回True
    """
    # 正常響應狀態碼：200
    res = response.response if response._status_code == 200 else None
    if res:
        res = json.loads(res[0])
        # 有響應數據，記一次
        return res.get("code") == 200

    return False


@api.route('/captcha')
@limit.limit("5/day;3/hour", deduct_when=func_deduct)
def expensive_route():
  ...

路由豁免：此情況特殊，屬于某個路由不參與頻率限制，使用方式為

limiter.exempt()

共享限制

適用于速率限制應由多條路由共享的情況。

命名共享限制

mysql_limit = limiter.shared_limit("100/hour", scope="mysql")

@app.route("..")
@mysql_limit
def r1():
   ...

@app.route("..")
@mysql_limit
def r2():
   ...

動態共享限制：將可調用對象作為范圍傳遞時，該函數的返回值將用作范圍。

def host_scope(endpoint_name):
    return request.host
host_limit = limiter.shared_limit("100/hour", scope=host_scope)

@app.route("..")
@host_limit
def r1():
   ...

@app.route("..")
@host_limit
def r2():
   ...

共享限制使用上與單個限制一致

配置

參數	說明
RATELIMIT_DEFAULT	默認策略, 逗號分隔('1/minute,100/hour')
RATELIMIT_DEFAULTS_PER_METHOD	是按方法/路線應用默認限制，還是按方法將所有方法組合應用默認限制。
RATELIMIT_DEFAULTS_EXEMPT_WHEN	默認豁免條件
RATELIMIT_APPLICATION	應用策略，用于將限制應用于整個應用程序（即，由所有路由共享）。
RATELIMIT_STORAGE_URL	存儲位置：

速度限制策略

Flask-Limiter內置了三種不同的速率限制策略。

分別為： Fixed Window、Fixed Window with Elastic Expiry、Moving Window

暫未研究，不做介紹。

錯誤響應

超出限制的請求返回的都是429狀態碼，示例如下：

<!DOCTYPE HTML PUBLIC "-//W3C//DTD HTML 3.2 Final//EN">
<title>429 Too Many Requests</title>
<h1>Too Many Requests</h1>
<p>1 per 1 day</p>

如果要配置響應，可對路由狀態碼判斷后響應，示例如下：

@app.errorhandler(429)
def ratelimit_handler(e):
    return make_response(
            jsonify(error="ratelimit exceeded %s" % e.description)
            , 429
    )

當然，還可以自定義錯誤信息：

app = Flask(__name__)
limiter = Limiter(app, key_func=get_remote_address)

def error_handler():
    return app.config.get("DEFAULT_ERROR_MESSAGE")

@app.route("/")
@limiter.limit("1/second", error_message='chill!')
def index():
    ....

@app.route("/ping")
@limiter.limit("10/second", error_message=error_handler)
def ping():
    ....

CBV與Blueprint使用

FBV可以使用裝飾器的方式進行限制，但是對于CBV就有些不適用了，以下就是CBV的使用方式。

app = Flask(__name__)
limiter = Limiter(app, key_func=get_remote_address)

class MyView(flask.views.MethodView):
    decorators = [limiter.limit("10/second")]
    def get(self):
        return "get"

    def put(self):
        return "put"

CBV的方式還是有些麻煩了，如果能對藍圖下所有的路由都進行限制就更好了，也可以對某個藍圖進行豁免。

app = Flask(__name__)
login = Blueprint("login", __name__, url_prefix = "/login")
regular = Blueprint("regular", __name__, url_prefix = "/regular")
doc = Blueprint("doc", __name__, url_prefix = "/doc")

@doc.route("/")
def doc_index():
    return "doc"

@regular.route("/")
def regular_index():
    return "regular"

@login.route("/")
def login_index():
    return "login"


limiter = Limiter(app, default_limits=["1/second"], key_func=get_remote_address)
limiter.limit("60/hour")(login)
limiter.exempt(doc)

app.register_blueprint(doc)
app.register_blueprint(login)
app.register_blueprint(regular)

關于代理

雖然上文說過Nginx代理的情況需要更復雜的操作，不過在查看官方文檔的時候，還發現了一個簡單的方法，說明如下：

如果您的應用程序位于代理之后，并且您使用的是werkzeug> 0.9+，則可以使用werkzeug.contrib.fixers.ProxyFix 修復程序可靠地獲取用戶的遠程地址，同時保護您的應用程序免于通過標頭進行ip欺騙。

from flask import Flask
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address
from werkzeug.contrib.fixers import ProxyFix

app = Flask(__name__)
# for example if the request goes through one proxy
# before hitting your application server
app.wsgi_app = ProxyFix(app.wsgi_app, num_proxies=1)
limiter = Limiter(app, key_func=get_remote_address)

API

flask_limit.Limiter類初始化屬性，Limiter(app=None, key_func=None, global_limits=[], default_limits=[], default_limits_per_method=False, default_limits_exempt_when=None, default_limits_deduct_when=None, application_limits=[], headers_enabled=False, strategy=None, storage_uri=None, storage_options={}, auto_check=True, swallow_errors=False, in_memory_fallback=[], in_memory_fallback_enabled=False, retry_after=None, key_prefix='', enabled=True)

參數	說明
app	即flask的項目
key_func	限制域
default_limits	默認限制策略
default_limits_per_method	默認限制是按方法/路線應用還是按每種方法所有方法的組合應用。
default_limits_exempt_when	默認豁免條件
default_limits_deduct_when	接收response對象并返回True / False以決定是否應從默認速率限制中扣除的函數
application_limits	所有路由的共享限制
headers_enabled	是否寫入響應頭
storage_uri	存儲位置
storage_options	意義不明的額外配置
auto_check	是否自動檢查應用程序的before_request鏈中的速率限制。默認`True`
swallow_errors	達到速率限制時會記錄異常。默認`False`
in_memory_fallback	字符串或可調用項的可變列表，返回表示存儲空間不足時要應用的回退限制的字符串
in_memory_fallback_enabled	僅在主存儲關閉并繼承原始限制時才退回到內存存儲中。
key_prefix	前綴
strategy	策略

方法：

check()

exempt()

ini_app()

request_filter()

reset()

limit(limit_value, key_func=None, per_method=False, methods=None, error_message=None, exempt_when=None, override_defaults=True, deduct_when=None)

shared_limit(limit_value, scope, key_func=None, error_message=None, exempt_when=None, override_defaults=True, deduct_when=None)

總結

以上是生活随笔為你收集整理的Flask-Limiter使用详细说明的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。