如何在结构化Flask应用中共享同一个Flask-Limiter实例
解决Flask拆分API后Flask-Limiter实例跨模块调用问题
方案1:将Limiter实例抽离到独立扩展模块
这是最清晰且易维护的方式,把全局扩展对象统一管理:
- 创建
extensions.py文件:
from flask_limiter import Limiter from flask_limiter.util import get_remote_address # 先初始化未绑定app的limiter实例 limiter = Limiter( key_func=get_remote_address, default_limits=["90 per minute"] )
- 修改
main.py,完成limiter与app的绑定:
from flask import Flask from werkzeug.middleware.proxy_fix import ProxyFix from apiv1 import blueprint as api1 from extensions import limiter REST_API = Flask(__name__) REST_API.wsgi_app = ProxyFix(REST_API.wsgi_app, x_for=1) # 绑定limiter到app,配置Redis存储(本地运行失败时自动降级) try: limiter.init_app( REST_API, storage_uri="redis://localhost:6379/1", strategy="fixed-window" ) except: limiter.init_app(REST_API) REST_API.register_blueprint(api1)
- 在
apiv1模块中直接导入扩展里的limiter使用:
from flask_restx import Namespace, Resource from extensions import limiter api = Namespace('apiv1', description='API v1 operations') @api.route('/example') class ExampleResource(Resource): @limiter.limit("30/minute") def get(self): return {"message": "Hello from API v1"}
方案2:通过蓝图传递Limiter实例
如果不想新增扩展模块,可以在注册蓝图时传入limiter实例:
- 修改
apiv1模块,改为工厂函数创建蓝图:
from flask import Blueprint from flask_restx import Api, Namespace, Resource def create_blueprint(limiter): blueprint = Blueprint('apiv1', __name__) api = Api(blueprint, version='1.0', title='API v1') ns = Namespace('example', description='Example operations') api.add_namespace(ns) @ns.route('/test') class TestResource(Resource): @limiter.limit("30/minute") def get(self): return {"data": "Test response"} return blueprint
- 修改
main.py的蓝图注册逻辑:
from flask import Flask from werkzeug.middleware.proxy_fix import ProxyFix from apiv1 import create_blueprint from flask_limiter import Limiter, get_remote_address REST_API = Flask(__name__) REST_API.wsgi_app = ProxyFix(REST_API.wsgi_app, x_for=1) # 初始化limiter try: limiter = Limiter( REST_API, key_func=get_remote_address, storage_uri="redis://localhost:6379/1", strategy="fixed-window", default_limits=["90 per minute"] ) except: limiter = Limiter( REST_API, key_func=get_remote_address, default_limits=["90 per minute"] ) # 传入limiter创建蓝图并注册 api1 = create_blueprint(limiter) REST_API.register_blueprint(api1)
方案3:通过current_app获取Limiter实例(备选)
利用Flask的current_app获取已注册的limiter实例,可读性较差,仅作为应急方案:
from flask import current_app from flask_restx import Namespace, Resource api = Namespace('apiv1', description='API v1 operations') @api.route('/demo') class DemoResource(Resource): @current_app.extensions['limiter'].limit("30/minute") def get(self): return {"message": "Demo response"}
内容的提问来源于stack exchange,提问作者Db0
相关产品推荐
相关产品推荐

