问题描述
这个问题是我上一个here的扩展。建议我多加解释。如标题所述，我试图找到一种避免将应用程序工厂（create_app函数）导入需要应用程序上下文并且“将current_app作为应用程序导入”的模块的方法。 >
我的问题是由于create_app这个函数，我需要通过循环导入才能获取app_context。
在我的__ini__.py中，我有这个：
# application/__init__.py
from flask import Flask
from flask_sqlalchemy import sqlAlchemy
from flask_restful import Api
from application.resources.product import Product,Products
from application.resources.offer import Offer,Offers # HERE IS THE PROBLEM
api = Api()
db = sqlAlchemy()
api.add_resource(Product,"/product/<string:name>") # GET,POST,DELETE,PUT to my local database
api.add_resource(Products,"/products") # GET all products from my local database
api.add_resource(Offer,"/offer/<int:id>") # POST call to the external Offers API microservise
api.add_resource(Offers,"/offers") # GET all offers from my local database
def create_app(config_filename=None):
""" Initialize core application. """
app = Flask(__name__,instance_relative_config=False)
app.config.from_object("config.Config")
db.init_app(app)
api.init_app(app)
with app.app_context():
db.create_all()
return app
问题出在这一行：
from application.resources.offer import Offer,Offers # HERE IS THE PROBLEM
因为在那个模块中，我有：
#application/resources/offer.py
from flask_restful import Resource
from application.models.offer import OfferModel # IMPORTING OFFER MODEL
依次导入 application / models / offer.py ，其中我有关键部分：
#application/models/offer.py
import requests
# from flask import current_app as app
from application import create_app # THIS CAUSES THE CIRculaR IMPORT ERROR
from sqlalchemy.exc import OperationalError
app = create_app() # I NEED TO CREATE THE APP IN ORDER TO GET THE APP CONTEXT BECASE IN THE CLASS I HAVE SOME FUNCTIONS THAT NEED IT
class OfferModel(db.Model):
""" Data model for offers. """
# some code to instantiate the class... + other methods..
# THIS IS ONE OF THE METHODS THAT NEED APP_CONTEXT OR ELSE IT WILL ERROR OUT
@classmethod
def update_offer_price(cls):
""" Call offers api to get new prices. This function will run in a separated thread in a scheduler. """
with app.app_context():
headers = {"Bearer": app.config["MS_API_ACCESS_TOKEN"]}
for offer_id in OfferModel.offer_ids:
offers_url = app.config["MS_API_OFFERS_BASE_URL"] + "/products/" + str(offer_id) + "/offers"
res = requests.get(offers_url,headers=headers).json()
for offer in res:
try:
OfferModel.query.filter_by(offer_id=offer["id"]).update(dict(price=offer["price"]))
db.session.commit()
except OperationalError:
print("Database does not exists.")
db.session.rollback()
我尝试使用from flask import current_app as app来获取上下文，但是它没有用。我不知道为什么不足以将current_app作为应用程序传递并获取上下文，因为它现在迫使我传递create_app应用程序工厂，这会导致循环导入问题。
解决方法
您的update_offer_price方法需要数据库交互和对配置的访问。它从应用程序上下文中获取它们，但仅在您的Flask应用程序初始化后才有效。此方法在单独的线程中运行，因此您在此线程中创建Flask应用程序的 second 实例。
另一种方法是在应用程序上下文之外获得独立的数据库交互和配置访问权限。
配置
配置似乎没有问题，因为您的应用程序是从另一个模块获取它的：
app.config.from_object("config.Config")
因此您可以直接将此对象导入到您的offer.py：
from config import Config
headers = {"Bearer": Config.MS_API_ACCESS_TOKEN}
数据库访问
要获得独立的数据库访问权限，您需要通过SQLAlchemy而非flask_sqlalchemy定义模型。 answer中已经对此进行了描述，但是我在此处发布了要点。对于您的情况，它可能看起来像这样。您的base.py模块：
from sqlalchemy import MetaData
from sqlalchemy.ext.declarative import declarative_base
metadata = MetaData()
Base = declarative_base(metadata=metadata)
和offer.py模块：
import sqlalchemy as sa
from .base import Base
class OfferModel(Base):
id = sa.Column(sa.Integer,primary_key=True)
# Another declarations
产生的metadata对象用于初始化flask_sqlalchemy对象：
from flask_sqlalchemy import SQLAlchemy
from application.models.base import metadata
db = SQLAlchemy(metadata=metadata)
可以在应用程序上下文之外查询您的模型，但是您需要手动创建数据库引擎和会话。例如：
from sqlalchemy import create_engine
from sqlalchemy.orm import Session
from config import Config
from application.models.offer import Offer
engine = create_engine(Config.YOUR_DATABASE_URL)
# It is recommended to create a single engine
# and use it afterwards to bind database sessions to.
# Perhaps `application.models.base` module
# is better to be used for this declaration.
def your_database_interaction():
session = Session(engine)
offers = session.query(Offer).all()
for offer in offers:
# Some update here
session.commit()
session.close()
请注意，通过这种方法，您不能使用模型类进行查询，我的意思是：
OfferModel.query.all() # Does not work
db.session.query(OfferModel).all() # Works
好的，这就是我解决的方法。我创建了一个新文件endpoints.py，在其中放置了所有的Api资源
# application/endpoints.py
from application import api
from application.resources.product import Product,Products
from application.resources.offer import Offer,Offers
api.add_resource(Product,"/product/<string:name>") # GET,POST,DELETE,PUT - calls to local database
api.add_resource(Products,"/products") # GET all products from local database.
api.add_resource(Offer,"/offer/<int:id>") # POST call to the Offers API microservice.
api.add_resource(Offers,"/offers") # GET all offers from local database
然后在init.py中，将其导入到最底部。
# aplication/__init__.py
from flask import Flask
from flask_restful import Api
from db import db
api = Api()
def create_app():
app = Flask(__name__,instance_relative_config=False)
app.config.from_object("config.Config")
db.init_app(app)
api.init_app(app)
with app.app_context():
from application import routes
db.create_all()
return app
from application import endpoints # importing here to avoid circular imports
它不是很漂亮，但是可以。