Flask-Limit使用詳細說明

時間  2020-07-11
標籤 flask limit 使用 詳細 說明 欄目 Python 简体版
原文   原文鏈接

Flask-Limit詳細說明

在flask項目中咱們須要對所有或者一部分接口進行限制，又不想造輪子，那怎麼辦呢？html

因此這就是flask-limit出現的緣由，不過對於相對複雜的需求，仍是本身造輪子吧！python

安裝與簡單使用

安裝：pip install Flask-Limitermysql

快速開始：redis

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"

上訴頻率限制說明：sql

  • 默認經過請求的remote_address進行限制。
  • 默認限制爲200次/天，50次/小時；適用於全部路線
  • slow路由的限制將繞過默認的速率限制，爲1次/天
  • medium路由繼承默認限制，並增長了1次/秒的限制
  • ping路由不受任何默認速率限制的約束

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

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

速率限制規則：flask

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

可使用本身選擇的分隔符將多個速率限制組合起來。api

示例：

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

使用的詳細說明

看完上面的部分其實已經知足大部分需求了，可是真實的狀況下，可能還存在其餘的定製服務，如下就是詳細說明。

初始化

初始化有兩種方式：

  1. 使用構造函數

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

limiter = Limiter(app, key_func=get_remote_address)

  2. 使用延遲應用初始化 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 存儲位置：
  • 內存：memcached://host:port
  • Redis： redis://host:port |
    | RATELIMIT_STORAGE_OPTIONS | 一個字典，用於設置要在初始化時傳遞給存儲實現的其餘選項。 |
    | RATELIMIT_STRATEGY | 使用的限速策略。詳見限速策略 |
    | RATELIMIT_HEADERS_ENABLED | 是否返回速率限制的相關信息到reponse header中。默認爲False，與上一條同樣能夠忽視。 |
    | RATELIMIT_ENABLED | 速率限制的整體終止開關。默認爲True |
    | RATELIMIT_HEADER_LIMIT | 當前速率限制的標題。默認爲X-RateLimit-Limit |
    | RATELIMIT_HEADER_RESET | 當前速率限制的重置時間的標題。默認爲X-RateLimit-Reset |
    | RATELIMIT_HEADER_REMAINING | 當前速率限制中剩餘的請求數的標頭。默認爲X-RateLimit-Remaining |
    | RATELIMIT_HEADER_RETRY_AFTER | 客戶端應什麼時候重試請求的標頭。默認爲Retry-After |
    | RATELIMIT_SWALLOW_ERRORS | 默認False便可 |
    | RATELIMIT_IN_MEMORY_FALLBACK_ENABLED | 若是啓用，則當配置的存儲關閉時，內存中的速率限制器將用做備用。與RATELIMIT_IN_MEMORY_FALLBACK原始速率限制結合使用時，將不會繼承該限制 |
    | RATELIMIT_IN_MEMORY_FALLBACK | 後端存儲異常使用的策略配置 |
    | RATELIMIT_KEY_PREFIX | 存儲key的前綴配置 |

速度限制策略

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)

相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程