Flask-应用程序范围的全局变量

Question

下午好。我正试图在Flask应用程序实例化期间创建一个"dpt_manager"单例对象，其中该管理器对象应该可以在整个应用程序和请求中访问。我知道应用程序上下文，我的应用程序工厂中创建的内容仅在请求/cli期间可用，但当收到请求时，我无法访问此变量(http://0.0.0.0:5000/getSampleResponse)。如果我的实现不理想，我也愿意学习更好的方法来实现烧瓶全局变量。

run.py:

from app import create_app
app = create_app()
if __name__ == "__main__":
app.run(debug=True, host="0.0.0.0", port=5000)

init.py:

from flask_api import FlaskAPI
from app.config import Config
from app.main.dpt_manager import DPTManager

def create_app():
app = FlaskAPI(__name__)
app.config.from_object(Config)
from app.api.casify_service.authorization import auth
from app.api.casify_service.predict import predict
app.register_blueprint(auth)
app.register_blueprint(predict)
# instantiate DPTManager instance
dpt_manager = DPTManager()  # <-- here is where I'm hoping to instantiate the object
return app

DPT管理员类别：

from app.main.data_set import DataSet
from app.main.pickle import Pickle
from app.config import Config

class DPTManager:
# TODO: singleton implementation goes here
def __init__(self):
self._model_retrain_freq = 30
self._is_model_persistent = True
def __str__(self):
return f"""
Model retrain frequency: {self._model_retrain_freq}
Model persistent state: {self._is_model_persistent}
"""

predict.py:

from flask import jsonify, current_app
from flask import Blueprint, request
from http import HTTPStatus
from app.util.auth import build_cors_preflight_response, corsify_response
from app.util.response import internal_server_error, successful_response
from app.util.decorators import token_required
import app.main.api_exceptions as err

predict = Blueprint("predict", __name__)    
@predict.route("/getSampleResponse", methods=["GET", "OPTIONS"])
@token_required
def get_sample_response():
"""Returns a sample response"""
if request.method == "GET":
sampleResponse = {"someKey": "someValue", "anotherKey": "anotherValue"}
print(current_app.dpt_manager) # <-- this doesn't work
# perform business logic here and build the response
response = jsonify(sampleData=sampleResponse)
return successful_response(response)

堆栈跟踪：

127.0.0.1 - - [28/Jun/2021 16:49:47] "GET /getSampleResponse HTTP/1.1" 500 -
Traceback (most recent call last):
File "/Users/user/Documents/environments/casify-service/lib/python3.9/site-packages/flask/app.py", line 1950, in full_dispatch_request
rv = self.dispatch_request()
File "/Users/user/Documents/environments/casify-service/lib/python3.9/site-packages/flask/app.py", line 1936, in dispatch_request
return self.view_functions[rule.endpoint](**req.view_args)
File "/Users/user/Documents/projects/casify_autoescalation/casify/casify-service/app/util/decorators.py", line 34, in decorated
return f(*args, **kwargs)
File "/Users/user/Documents/projects/casify_autoescalation/casify/casify-service/app/api/casify_service/predict.py", line 23, in get_sample_response
print(current_app.dpt_manager)
File "/Users/user/Documents/environments/casify-service/lib/python3.9/site-packages/werkzeug/local.py", line 347, in __getattr__
return getattr(self._get_current_object(), name)
AttributeError: 'FlaskAPI' object has no attribute 'dpt_manager'

Answer 1

我建议使用新文件extensions.py

1. 在应用程序文件夹下创建新文件，将其命名为extensions.py

from app.main.dpt_manager import DPTManager

dpt_manager = DPTManager()# init the instance here so we can import later 

2. init.py

from flask_api import FlaskAPI

from app.config import Config
from app.extensions import dpt_manager 
def create_app():
app = FlaskAPI(__name__)
app.config.from_object(Config)
from app.api.casify_service.authorization import auth
from app.api.casify_service.predict import predict
app.register_blueprint(auth)
app.register_blueprint(predict)
print(dpt_manager) ## let print out to check
return app

3. 然后在predict.py中，让导入dpt_manager并使用它

from flask import jsonify, current_app
from flask import Blueprint, request
from http import HTTPStatus
from app.util.auth import build_cors_preflight_response, corsify_response
from app.util.response import internal_server_error, successful_response
from app.util.decorators import token_required
from app.extensions import dpt_manager 
import app.main.api_exceptions as err

predict = Blueprint("predict", __name__)    
@predict.route("/getSampleResponse", methods=["GET", "OPTIONS"])
@token_required
def get_sample_response():
"""Returns a sample response"""
if request.method == "GET":
sampleResponse = {"someKey": "someValue", "anotherKey": "anotherValue"}
print(dpt_manager) # your instance 
# perform business logic here and build the response
response = jsonify(sampleData=sampleResponse)
return successful_response(response)

Answer 2

您可以尝试将需要全局访问的对象存储在会话数据中，会话数据在语法上类似于Python字典。

存储：

from flask import session
session['dpt_manager'] = dpt_manager

在其他地方检索：

from flask import session
dpt_manager = session.get('dpt_manager') 

请注意，如果"dpt_manager"不在会话中，session.get('dpt_manager')将返回None，与其立即使用会话"dpt_manager"，不如立即使用