避免将应用程序工厂导入需要应用程序上下文的模块

Question

这个问题是对我之前的问题here 的扩展。有人建议我多放一点来解释这个问题。正如标题所说，我试图找到一种方法来避免将应用程序工厂（create_app 函数）导入需要应用程序上下文的模块中，并且“将 current_app 作为应用程序导入”是不够的。 p>

我的问题是我有一个循环导入问题，因为我需要传递这个 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 应用程序工厂。

Answer 1

您的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

Answer 2

好的，这就是我解决它的方法。我创建了一个新文件 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

它不是很漂亮，但它可以工作。