Flask-Limiter详细使用说明

本文首发于：行者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】，与行者一起讨论吧！