本文首发于：行者AI
速率限制通常作为服务的防御措施予以实施。服务需要保护自身以免过度使用（无论是有意还是无意），从而保持服务可用性。在Flask项目开发过程中，遇到了需要对接口进行限制的需求，又不想去造轮子，这时候就需要用到Flask-Limiter这个三方库。本文将对Flask-Limiter的使用进行详细说明。
1. 安装
安装依赖环境。
pip install Flask==1.1.1 Flask-Limiter==1.4
2. 快速开始
有两种方式表示速率限制：
- "100 per day"、"20 per hour"、"5 per minute"、"1 per second"
- "100/day"、"20/hour"、"5/minute"、"1/second"
速率限制可以设置全局配置，针对所有接口进行限制；也可以通过装饰器进行局部限制；对于不想限制的接口，可以通过装饰器@limiter.exempt进行解除限制。示例代码如下所示：
app = Flask(__name__)
# 该配置为全局配置、适用于所有接口
limiter = Limiter(app, key_func=get_remote_address, default_limits=["100 per day", "10/hour"])
# @limiter.limit: 将覆盖全局limiter配置
@app.route("/slow")
@limiter.limit("1 per day")
def slow():
return ":("
# override_defaults: 表示该limiter是否覆盖全局limiter限制，默认为True
@app.route("/medium")
@limiter.limit("1/second", override_defaults=False)
def medium():
return ":|"
# 完整继承全局limiter配置
@app.route("/fast")
def fast():
return ":)"
# @limiter.exempt: 被装饰的视图不受全局速率限制
@app.route("/ping")
@limiter.exempt
def ping():
return "PONG"
3. 装饰器
根据个人喜好和使用场景，有以下几种方式：
单一修饰：限制字符串可以是单个限制，也可以是定界符分隔的字符串。
@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():
...
自定义功能：默认情况下，根据Limiter实例初始化时所使用的关键功能来应用速率限制。开发者可以实现自己的功能。
def my_key_func():
...
@app.route("...")
@limiter.limit("100/day", my_key_func)
def my_route():
...
动态加载限制的字符串：在某些情况下，需要从代码外部的源（数据库，远程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():
...
4. 限制域
指根据什么进行限制，对应的参数为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服务器的地址，相当于来自该Nginx服务器的所有请求会被当做同一个IP访问，所以项目中一般都是自定义key_func！
def limit_key_func():
return str(flask_request.headers.get("X-Forwarded-For", '127.0.0.1'))
获取IP的依据是根据Nginx的配置决定的：
X-Forwarded-For # 一般是每一个非透明代理转发请求时会将上游服务器的ip地址追加到X-Forwarded-For的后面，使用英文逗号分割；
X-Real-IP # 一般是最后一级代理将上游ip地址添加到该头中；
X-Forwarded-For # 是多个ip地址，而X-Real-IP是一个；
# 如果只有一层代理，这两个头的值就是一样的。
5. 共享限制
适用于速率限制由多条路由共享的情况。
命名共享限制：通过相同的shared_limit对象进行装饰。
mysql_limit = limiter.shared_limit("100/hour", scope="mysql")
@app.route("..")
@mysql_limit
def r1():
...
@app.route("..")
@mysql_limit
def r2():
...
动态共享限制：将可调用对象作为范围传递，该函数的返回值将用作范围。注意，callable具有一个参数：表示请求端点的字符串。
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():
...
6. 储存后端
记录IP访问次数，用于判断该IP访问次数是否达到限制。Limiter默认使用内存作为储存后端，但是在实际开发中，可能会涉及到多进程资源不共享、服务器内存消耗等问题，一般是使用redis作为储存后端。
-
需要redis服务器的位置，以及可选的数据库号。
redis://localhost:6379或redis://localhost:6379/n（对于数据库n）。 -
如果redis服务器正在通过unix域套接字监听，则可以使用
redis+unix:///path/to/sock或redis+unix:///path/to/socket?db=n（对于数据库n）。 -
如果数据库受密码保护，则可以在URL中提供密码，例如，
redis://:foobared@localhost:6379或者redis+unix//:foobered/path/to/socket。
# LIMITS_REDIS_STORAGE就是上文提到的redis的URI
limiter = Limiter(default_limits=["100 per day"], key_func=limit_key_func, storage_uri=LIMITS_REDIS_STORAGE)
7. 总结
通过对API请求的限制和配额，在一定程度上能避免系统收到超出其处理能力的数据，确保系统的资源能得到合理的使用。以上就是本文的全部内容，希望能给大家学习带来帮助。
PS：更多技术干货，快关注【公众号 | xingzhe_ai】，与行者一起讨论吧！