Flask-limiter修改错误响应码

flask-limiter文档：https://flask-limiter.readthedocs.io/en/stable/

初始化

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)

3、自定义键控函数

通过当前用户限制路由的速率(使用Flask-Login):

@route("/test")
@login_required
@limiter.limit("1 per day", key_func = lambda : current_user.username)
def test_route():
    return "42"

按地区名限制所有请求的费率:

from flask import request, Flask
import GeoIP
gi = GeoIP.open("GeoLiteCity.dat", GeoIP.GEOIP_INDEX_CACHE | GeoIP.GEOIP_CHECK_CACHE)

def get_request_country():
    return gi.record_by_name(request.remote_addr)['region_name']

app = Flask(__name__)
limiter = Limiter(app, default_limits=["10/hour"], key_func = get_request_country)

key_func是从 flask request context中调用的。该函数可实现根据IP、用户等字段进行限定。

使用限定

FBV方式（装饰器）：

1、单装饰

限制字符串可以是单个限制或分隔符分隔的字符串。

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

2、多装饰

限制字符串可以是单个限制或分隔符分隔的字符串，也可以是两者的组合。

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

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"

Blueprint：

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)

自定义速率限制超过响应

默认配置导致抛出一个 RateLimitExceeded异常(这会有效地停止任何进一步处理和状态为“429”的响应)。

所有的频率限定：

方式一：

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

方式二：

from flask import make_response, render_template
from flask_limiter import Limiter, RequestLimit

def default_error_responder(request_limit: RequestLimit):
    return make_response(
        render_template("my_ratelimit_template.tmpl", request_limit=request_limit)
        429
    )

app = Limiter(
    key_func=...,
    default_limits=["100/minute"],
    on_breach=default_error_responder
)

方式三：

修改flask_limiter库中errors.py中的RateLimitExceeded类

注：该方式用于修改CBV且flask_limiter的版本较老，Limiter类中没有on_breach参数

class RateLimitExceeded(werkzeug_exception):
    """exception raised when a rate limit is hit.

    The exception results in ``abort(429)`` being called.
    """

    code = 429
    limit = None

    def __init__(self, limit):
        self.limit = limit
        if limit.error_message:
            description = (
                limit.error_message
                if not callable(limit.error_message)
                else limit.error_message()
            )
        else:
            description = text_type(limit.limit)
        super(RateLimitExceeded, self).__init__(description=description)

改成：

class RateLimitExceeded(werkzeug_exception):
    """exception raised when a rate limit is hit.

    The exception results in ``abort(429)`` being called.
    """

    code = 200
    limit = None

    def __init__(self, limit):
        self.limit = limit
        if limit.error_message:
            description = (
                limit.error_message
                if not callable(limit.error_message)
                else limit.error_message()
            )
        else:
            description = text_type(limit.limit)
        super(RateLimitExceeded, self).__init__(description=description)

该处429可以改成200，网站的请求会返回200响应

自定义限制请求头

可以从request context中的Limiter实例访问current_limit属性。

例如：你可以在内置的header中使用after_request()装饰器添加自己的header填充:

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


@app.route("/")
@limiter.limit("1/second")
def index():
    ....

@app.after_request
def add_headers(response):
    if limiter.current_limit:
        response.headers["RemainingLimit"] = limiter.current_limit.remaining
        response.headers["ResetAt"] = limiter.current_limit.reset_at
        response.headers["MaxRequests"] = limiter.current_limit.limit.amount
        response.headers["WindowSize"] = limiter.current_limit.limit.get_expiry()
        response.headers["Breached"] = limiter.current_limit.breached
    return response

和代理一起使用

werkzeug

如果应用程序在代理的后面，并且正在使用werkzeug > 0.9+，可以使用werkzeug.middleware.proxy_fix获取用户的远程地址，同时保护应用程序免受通过头文件的ip欺骗。

from flask import Flask
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address
from werkzeug.middleware.proxy_fix 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, x_for=1)
limiter = Limiter(app, key_func=get_remote_address)

nginx

1、nginx配置修改加上

proxy_set_header X-Real-IP $remote_addr;

2、获取ip地址函数修改

def get_remote_address():
    """
    :return: the ip address for the current request (or 127.0.0.1 if none found)
    """
    return request.remote_addr or '127.0.0.1'

改成：

def get_remote_address():
    """
    :return: the ip address for the current request (or 127.0.0.1 if none found)
    """
	return request.headers.get("X-Real-IP")

重启nginx

nginx -s reload