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

Question

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

运行.py：

from app import create_app

app = create_app()

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

初始化.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

DPTManager 类：

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. 在app文件夹下新建文件，命名为extensions.py

from app.main.dpt_manager import DPTManager
        
dpt_manager = DPTManager()# init the instance here so we can import later 

1. 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

1. 然后在您的 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，最好立即使用它而不是 session['dpt_manager']