标签：远程 lang port mes 规则 设置 路由 命名 install
在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进行限制。注意： 静态路由不受速率限制
每次请求超出速率限制时，将不会调用view函数，而是会引发429http错误。
速率限制规则：
[count] [per|/] [n (optional)] [second|minute|hour|day|month|year]
可以使用自己选择的分隔符将多个速率限制组合起来。
示例：
看完上面的部分其实已经满足大部分需求了，但是真实的情况下，可能还存在其他的定制服务，以下就是详细说明。
初始化有两种方式：
使用构造函数
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提供了两种方式：
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 | 存储位置： |
False，与上一条一样可以忽视。 |True |X-RateLimit-Limit |X-RateLimit-Reset |X-RateLimit-Remaining |Retry-After |RATELIMIT_IN_MEMORY_FALLBACK原始速率限制结合使用时，将不会继承该限制 |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():
....
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)
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)
标签：远程 lang port mes 规则 设置 路由 命名 install
原文地址：https://www.cnblogs.com/Du704/p/13281032.html
