速率限制通常作为服务的防御措施予以实施。服务需要保护自身以免过度使用（无论是有意还是无意），从而保持服务可用性。在Flask项目开发过程中，遇到了需要对接口进行限制的需求，又不想去造轮子，这时候就需要用到Flask-Limiter这个三方库。

安装与简单使用

安装：pip install Flask-Limiter

快速开始：

有两种方式表示速率限制：

• "100 per day"、"20 per hour"、"5 per minute"、"1 per second"
• "100/day"、"20/hour"、"5/minute"、"1/second"

速率限制可以设置全局配置，针对所有接口进行限制；也可以通过装饰器进行局部限制；对于不想限制的接口，可以通过装饰器@limiter.exempt进行解除限制。示例代码如下所示：

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"]
)

# slow路由的限制将绕过默认的速率限制，为1次/天
@app.route("/slow")
@limiter.limit("1 per day")
def slow():
    return ":("

# @limiter.exempt: 被装饰的视图不受全局速率限制
@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"

上诉频率限制说明：

• 默认通过请求的remote_address进行限制。
• 默认限制为200次/天，50次/小时；适用于所有路线
• slow路由的限制将绕过默认的速率限制，为1次/天
• medium路由继承默认限制，并增加了1次/秒的限制
• ping路由不受任何默认速率限制的约束

注意： 静态路由不受速率限制

每次请求超出速率限制时，将不会调用view函数，而是会引发429http错误。

速率限制规则：

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

可以使用自己选择的分隔符将多个速率限制组合起来。

示例：

• 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():

  ...

共享限制使用上与单个限制一致

配置

• 内存：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代理的情况需要更复杂的操作，不过在查看官方文档的时候，还发现了一个简单的方法，说明如下：

0.9+，则可以使用werkzeug.contrib.fixers.ProxyFix 修复程序可靠地获取用户的远程地址，同时保护您的应用程序免于通过标头进行ip欺骗。

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)

方法：

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)

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)

本栏目推荐文章
• PasteSpider之接口的授权实现为什么不采用JWT方式
• Dating Java8系列之Lambda表达式和函数式接口（上）
• Dating Java8系列之Lambda表达式和函数式接口（下）
• limit 影响性能的原因和优化方案
• 通过API接口操作vmware vcenter虚拟机
• TVBox等智能电视第三方APP可用直播源接口【截止24.01.12】
• 应用层限流——四种接口限流算法原理及实现
• 自动生成接口文档
• 常见的HTTP接口超时问题出现原因及解决办法
• 03.接口测试用例设计

Flask-Limit 接口 Flask Limitflask-limit接口flask limit flask-limit plugin-limit插件 框架 接口 paddledetection接口flask 接口 概念flask api mail_api_flask接口 项目flask multiprocess接口gevent flask 接口 功能python flask 接口 文件flask api limit