Flask 速率限制器不适用于 Flask-restful

Question

我相信我遇到了与此处描述的相同的问题：Flask-Limiter 不适用于基于 Flask-Restful API 的应用程序。但我仍然无法让它发挥作用。这是我的问题：

• 我有一个

  app.py

  ，我在其中导入所有 REST-API 端点。我使用本指南中所述的 Flask-RESTful 资源：https://flask-restful.readthedocs.io/en/latest/quickstart.html

• 我使用“Flask-Limiter”库进行速率限制：https://flask-limiter.readthedocs.io/en/stable/。速率限制器在单独的文件中实例化，而不是在

  app.py

  内，因为这会在

  app.py

  和 REST-API 端点之间创建循环引用。我认为这可能是我的问题的原因。

这是速率限制器文件（位于：

/controller/helpers/rate_limiter.py

）

from flask import current_app
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address

limiter = Limiter(
    current_app,
    key_func=get_remote_address
)

这是

app.py

文件（位于：

/

）：

from flask import Flask
from flask_restful import Api
from controller.actions.my_endpoint import MyEndpoint

flask_app = Flask(__name__)
api = Api(flask_app)
api.add_resource(MyEndpoint, '/endpoint')
flask_app.run(debug=True, use_reloader=True, host='0.0.0.0', port=8000)

最后这是一个端点的示例（位于：

/controller/actions/my_endpoint

）：

from flask_restful import reqparse, Resource
from controller.helpers.rate_limiter import limiter


class MyEndpoint(Resource):
    decorators = [limiter.limit("1/minute", methods=["POST"])]
    
    def __init__(self):
        self.parser = reqparse.RequestParser()
        self.parser.add_argument(
            "someData", type=int, nullable=False, required=True)

    def post(self):
        data = self.parser.parse_args(strict=True)
        someData = data.someData

        return "Got your message {0}".format(someData), 200

我希望在第一个请求之后我会得到响应 429，因为达到了 1/分钟的速率限制，但它并没有这样做。

但是，当我在

app.py

中实例化限制器并设置默认速率限制时，它会起作用（出现错误 429）：

from flask import Flask
from flask_restful import Api
from controller.db_actions.my_endpoint import MyEndpoint
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address

flask_app = Flask(__name__)

limiter = Limiter(
    flask_app,
    key_func=get_remote_address,
    default_limits=["1 per minute"]
)

api = Api(flask_app)
api.add_resource(MyEndpoint, '/endpoint')
flask_app.run(debug=True, use_reloader=True, host='0.0.0.0', port=8000)

问题：我认为我在实例化限制器时做错了什么。它是什么？ :)

Answer 1

我通过使用 utils.py 文件并从中导入限制器来解决这个问题。

在utils.py中

from flask_limiter import Limiter
from flask_limiter.util import get_remote_address

# Setup the Rate Limiter
limiter = Limiter(
    get_remote_address,
    default_limits=["100 per hour", "10 per minute"],
    storage_uri="memory://",
)

然后我导入它并在限制器上使用

init_app()

方法使其按预期工作。

在

__init__.py

from .utils import limiter

app = Flask(__name__, static_folder="static", static_url_path="/static")

limiter.init_app(app)

现在我可以将其导入到我的蓝图中并且它可以工作：

from rest_api.utils import limiter

@employee_blueprint.route("/")
@limiter.limit("1 per second")
def get_employees(session=""):
   ... code here ...

我受到这篇文章的启发。