最新消息:20210816 当前crifan.com域名已被污染,为防止失联,请关注(页面右下角的)公众号

【已解决】Flask中如何在其他模块中引入当前的app或者全局单一实例的app

Flask crifan 278浏览 0评论
折腾:
【未解决】Flask中循环导入app和create_app的问题:ImportError: cannot import name ‘create_app’
期间,为了解决Flask中app,create_app中循环导入的问题,需求是:
可以在其他模块中,导入app,以及初始化好的log,然后后续调用app.config
现在是直接用:
resources/tts.py
from app import app, log
希望改为:
引用全局的app,记得是current_app?
以及用flask中的g,去保存相关其他全局变量?
或者是用app.logger,即可获取log
最好是,多个app实例,也同时共享一个app
不过还是先去解决循环引用app的问题
flask other module import current app
python – How to share the global app object in flask? – Stack Overflow
from flask import current_app as app
感觉不错
去试试,好像的确是可以的,不会再去导入(此处正在初始化的app),而是可以继续下一步执行后续代码了:
不过后续代码执行,发现此处app不是实例,而是
<LocalProxy unbound>
而且还无法解析到值:
然后报错了:
Error evaluating: thread_id: pid_72249_id_4349874640
frame_id: 4379058248
scope: FRAME
attrs: app
Traceback (most recent call last):
  File "/Applications/PyCharm.app/Contents/helpers/pydev/_pydevd_bundle/pydevd_vars.py", line 248, in resolve_compound_variable_fields
    return _typeName, resolver.get_dictionary(var)
  File "/Applications/PyCharm.app/Contents/helpers/pydev/_pydevd_bundle/pydevd_resolver.py", line 75, in get_dictionary
    return self._getPyDictionary(var, names)
  File "/Applications/PyCharm.app/Contents/helpers/pydev/_pydevd_bundle/pydevd_resolver.py", line 148, in _getPyDictionary
    names = self.get_names(var)
  File "/Applications/PyCharm.app/Contents/helpers/pydev/_pydevd_bundle/pydevd_resolver.py", line 137, in get_names
    if not names and hasattr(var, '__members__'):
  File "/Users/crifan/.virtualenvs/xxxRobotDemoServer-SCpLPEyZ/lib/python3.6/site-packages/werkzeug/local.py", line 346, in __getattr__
    return dir(self._get_current_object())
  File "/Users/crifan/.virtualenvs/xxxRobotDemoServer-SCpLPEyZ/lib/python3.6/site-packages/werkzeug/local.py", line 306, in _get_current_object
    return self.__local()
  File "/Users/crifan/.virtualenvs/xxxRobotDemoServer-SCpLPEyZ/lib/python3.6/site-packages/flask/globals.py", line 51, in _find_app
    raise RuntimeError(_app_ctx_err_msg)
RuntimeError: Working outside of application context.

This typically means that you attempted to use functionality that needed
to interface with the current application object in some way. To solve
this, set up an application context with app.app_context().  See the
documentation for more information.

python – Import module in Flask – Stack Overflow
python – Reuse app object in Flask – Stack Overflow
python – current_app and importing config settings in Flask – Stack Overflow
The Application Context — Flask 1.0.2 documentation
“Purpose of the Context
The Flask application object has attributes, such as config, that are useful to access within views and CLI commands. However, importing the app instance within the modules in your project is prone to circular import issues. When using the app factory pattern or writing reusable blueprints or extensions there won’t be an app instance to import at all. “
此处的current_app,就是为了避免这个循环引用的问题的
-》所以其他模块,如果想要引用app,不要用:
from app import app
以及我这里的:
from app import app, log
而应该改为:
from flask import current_app as app

def someFunction():
    log = app.logger
    log.info("xxx")
即可。
然后也清楚此处的current_app,就像PyCharm调试期间看到的,是个<LocalProxy unbound>
是因为:
还没有进入Flask的views等环境中,
如果外界访问一个api,然后调用进入某个view的get/put/post等函数中,自然就可以获取对应的app(以及request的context
-》对应的,当request结束时,就弹出取消掉current_app(和request的context)了。
而上述办法,对每个函数都要添加:
log = app.logger
觉得很繁琐,所以又期间尝试简化去用:
# log = app.logger
log = None
with app.app_context():
    print("in flask app context, so init log from app.logger")
    log = app.logger
结果还是会报错:
  File "/Users/crifan/dev/dev_root/company/xxx/projects/robotDemo/server/xxxRobotDemoServer/resources/tts.py", line 54, in <module>
    with app.app_context():
  File "/Users/crifan/.virtualenvs/xxxRobotDemoServer-SCpLPEyZ/lib/python3.6/site-packages/werkzeug/local.py", line 347, in __getattr__
    return getattr(self._get_current_object(), name)
  File "/Users/crifan/.virtualenvs/xxxRobotDemoServer-SCpLPEyZ/lib/python3.6/site-packages/werkzeug/local.py", line 306, in _get_current_object
    return self.__local()
  File "/Users/crifan/.virtualenvs/xxxRobotDemoServer-SCpLPEyZ/lib/python3.6/site-packages/flask/globals.py", line 51, in _find_app
    raise RuntimeError(_app_ctx_err_msg)
RuntimeError: Working outside of application context.

This typically means that you attempted to use functionality that needed
to interface with the current application object in some way. To solve
this, set up an application context with app.app_context().  See the
documentation for more information.
不过后来用:
def create_rest_api(app):
    from resources.qa import RobotQaAPI
    from resources.asr import RobotAsrAPI
    from resources.files import GridfsAPI, TmpAudioAPI

    rest_api = Api()
    rest_api.add_resource(RobotQaAPI, '/qa', endpoint='qa')
    rest_api.add_resource(RobotAsrAPI, '/asr/language/<string:language>', endpoint='asr')
    rest_api.add_resource(GridfsAPI, '/files/<fileId>', '/files/<fileId>/<fileName>', endpoint='gridfs')
    rest_api.add_resource(TmpAudioAPI, '/tmp/audio/<filename>', endpoint='TmpAudio')

    rest_api.init_app(app)
    return rest_api

################################################################################
# Global Init App
################################################################################

print("in flask app: settings=%s" % (settings))
app = create_app(settings)
# register_extensions(app)
log = app.logger
log.debug("app=%s", app)

log.debug("log=%s", log)
log.debug("settings.FLASK_ENV=%s", settings.FLASK_ENV)
log.debug("settings.DEBUG=%s, settings.MONGODB_HOST=%s, settings.FILE_URL_HOST=%s",
          settings.DEBUG, settings.MONGODB_HOST, settings.FILE_URL_HOST)

with app.app_context():
    # api = Api(app)
    api = Api()  # <flask_restful.Api object at 0x1064e5438>
    log.debug("api=%s", api)
    api = create_rest_api(app)
    log.debug("api=%s", api)
至少可以避免app.py中的上述报错。
但是:
【已解决】Flask中已经用了with app.app_context()调用其他模块时g出错:AttributeError: ‘_AppCtxGlobals’ object has no attribute ‘app’
【总结】
至此,算是基本上解决了,通过使用g解决了全局变量调用,避免了循环导入的问题:
(1)app中只是导入factory的create_app
app.py
from conf.app import settings

from factory import create_app
print("in flask app: settings=%s" % (settings))
app = create_app(settings)
# register_extensions(app)
log = app.logger
log.debug("app=%s", app)

log.debug("log=%s", log)
log.debug("settings.FLASK_ENV=%s", settings.FLASK_ENV)
log.debug("settings.DEBUG=%s, settings.MONGODB_HOST=%s, settings.FILE_URL_HOST=%s",
          settings.DEBUG, settings.MONGODB_HOST, settings.FILE_URL_HOST)

if __name__ == "__main__":
    app.run(
        host=app.config["FLASK_HOST"],
        port=app.config["FLASK_PORT"],
        debug=app.config["DEBUG"]
    )
(2)factory中导入g,并在create_app加上with app.app_context(),然后对于初始化各种extension后,赋值给g
factory.py
import os
from flask import Flask

import logging
from logging.handlers import RotatingFileHandler

# from flask_pymongo import PyMongo
from gridfs import GridFS
from pymongo import MongoClient

from flask_restful import Api
from flask_cors import CORS
from celery import Celery

from flask import g

################################################################################
# Global Variables
################################################################################
# # log = logging.getLogger() #<RootLogger root (WARNING)>
# log = None
# print("log=%s" % log)
#
# # mongo = MongoClient() # MongoClient(host=['localhost:27017'], document_class=dict, tz_aware=False, connect=True)
# mongo = None
# print("mongo=%s" % mongo)
# fsCollection = None #None
# print("fsCollection=%s" % fsCollection)
#
# # celery = Celery() #<Celery __main__ at 0x1068d8b38>
# celery = None
# print("celery=%s" % celery)

################################################################################
# Global Function
################################################################################

def create_app(config_object):
    # global log

    # app = Flask(__name__) #<Flask 'factory'>
    app = Flask(config_object.FLASK_APP_NAME) #<Flask 'RobotQA'>
    CORS(app)

    # app.config.from_object('config.DevelopmentConfig')
    # # app.config.from_object('config.ProductionConfig')

    app.config.from_object(config_object)
    with app.app_context():
        g.app = app

        log = create_log(app)
        g.log = log

        log.debug("after load from object: app.config=%s", app.config)
        log.debug('app.config["DEBUG"]=%s, app.config["MONGODB_HOST"]=%s, app.config["FILE_URL_HOST"]=%s',
                  app.config["DEBUG"], app.config["MONGODB_HOST"], app.config["FILE_URL_HOST"])

        register_extensions(app)
        log.info("flask app extensions init completed")

    return app


def register_extensions(app):
    # global log, mongo, fsCollection, api, celery

    log = g.log

    mongo = create_mongo(app)
    g.mongo = mongo
    log.info("mongo=%s", mongo)
    mongoServerInfo = mongo.server_info()
    log.debug("mongoServerInfo=%s", mongoServerInfo)

    fsCollection = create_gridfs_fs_collection(mongo)
    g.fsCollection = fsCollection
    log.info("fsCollection=%s", fsCollection)

    celery = create_celery(app)
    g.celery = celery
    log.info("celery=%s", celery)

    # api = Api(app)
    api = create_rest_api(app)
    log.debug("api=%s", api)
    g.api = api

    return app


def create_rest_api(app):
    from resources.qa import RobotQaAPI
    from resources.asr import RobotAsrAPI
    from resources.files import GridfsAPI, TmpAudioAPI

    rest_api = Api()
    rest_api.add_resource(RobotQaAPI, '/qa', endpoint='qa')
    rest_api.add_resource(RobotAsrAPI, '/asr/language/<string:language>', endpoint='asr')
    rest_api.add_resource(GridfsAPI, '/files/<fileId>', '/files/<fileId>/<fileName>', endpoint='gridfs')
    rest_api.add_resource(TmpAudioAPI, '/tmp/audio/<filename>', endpoint='TmpAudio')

    rest_api.init_app(app)
    return rest_api


def create_log(app):
    print("create_log: before init log: app.logger=%s" % app.logger)
    logFormatterStr = app.config["LOG_FORMAT"]
    logFormatter = logging.Formatter(logFormatterStr)

    fileHandler = RotatingFileHandler(
        app.config['LOG_FILE_FILENAME'],
        maxBytes=app.config["LOG_FILE_MAX_BYTES"],
        backupCount=app.config["LOG_FILE_BACKUP_COUNT"],
        encoding="UTF-8")
    fileHandler.setLevel(logging.DEBUG)
    fileHandler.setFormatter(logFormatter)
    app.logger.addHandler(fileHandler)
    # Note: should NOT set StreamHandler here, otherwise will duplicate debug log
    app.logger.setLevel(logging.DEBUG)  # set root log level

    log = app.logger
    log.info("app=%s", app)
    # log.debug("app.config=%s", app.config)
    print("create_log: after init log: app.logger=%s" % app.logger)

    return log


def create_celery(app):
    celery = Celery(app.name, broker=app.config['CELERY_BROKER_URL'])
    celery.conf.update(app.config)

    TaskBase = celery.Task

    class ContextTask(TaskBase):
        abstract = True

        def __call__(self, *args, **kwargs):
            with app.app_context():
                g.log.info("in celery ContextTask __call__: args=%s, kwargs=%s", args, kwargs)
                return TaskBase.__call__(self, *args, **kwargs)

    celery.Task = ContextTask

    return celery


def create_mongo(app):
    mongo_client = MongoClient(
        host=app.config["MONGODB_HOST"],
        port=app.config["MONGODB_PORT"],
        username=app.config["MONGODB_USERNAME"],
        password=app.config["MONGODB_PASSWORD"],
        authSource=app.config["MONGODB_AUTH_SOURCE"]
    )
    return mongo_client


def create_gridfs_fs_collection(mongo_db):
    # Pure PyMongo
    gridfs_db = mongo_db.gridfs  # Database(MongoClient(host=['47.x.x.x:22222'], document_class=dict, tz_aware=False, connect=True, authsource='gridfs'), 'gridfs')
    gridfs_fs_collection = GridFS(gridfs_db)  # <gridfs.GridFS object at 0x1107b2390>
    return gridfs_fs_collection
(3)对于factory中的初始化flask-restful的Api时,所引用了别的资源模块,比如:
from resources.qa import RobotQaAPI
rest_api.add_resource(RobotQaAPI, ‘/qa’, endpoint=’qa’)
其中就可以导入g,并从g中得到初始化后的,全局的log,app等其他实例了
resources/qa.py
from flask import g

app = g.app
log = g.log

class RobotQaAPI(Resource):

    def processResponse(self,
                        respDict,
                        voiceName=MS_TTS_VOICE_NAME,
                        voiceRate=MS_TTS_VOICE_RATE,
                        voiceVolume=MS_TTS_VOICE_VOLUME):
        """
            process response dict before return
                generate audio for response text part
        """
        # log = app.logger
        unicodeText = respDict["data"]["response"]["text"]
        log.info("unicodeText=%s")

转载请注明:在路上 » 【已解决】Flask中如何在其他模块中引入当前的app或者全局单一实例的app

发表我的评论
取消评论

表情

Hi,您需要填写昵称和邮箱!

  • 昵称 (必填)
  • 邮箱 (必填)
  • 网址
86 queries in 0.113 seconds, using 20.51MB memory