这个问题是我上一个问题的扩展。有人建议我多放些东西来解释这个问题。正如标题所说，我试图找到一种方法来避免将应用程序工厂(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
这反过来又导入应用程序/模型/报价.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应用程序的第二个实例。
另一种方法是在应用程序上下文之外获得独立的数据库交互和配置访问。
配置
配置似乎不是问题，因为您的应用程序是从另一个模块获得的：
app.config.from_object("config.Config")
因此，您可以直接将此对象导入您的offer.py:
from config import Config
headers = {"Bearer": Config.MS_API_ACCESS_TOKEN}
数据库访问
要获得独立的数据库访问权限，您需要通过SQLAlchemy而不是flask_sqlalchemy定义模型。这个答案已经描述过了，但我在这里发布了要点。就你的情况而言，它可能是这样的。您的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
它不是很漂亮，但很管用。