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

Question

这个问题是我上一个问题的扩展。有人建议我多说一些来解释这个问题。正如标题所说，我试图找到一种方法来避免将应用程序工厂（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 作为 app 传递并获取上下文是不够的，因为它现在迫使我传递导致循环导入问题的 create_app 应用程序工厂。

标签： pythonapiflaskflask-restful

Answer 1

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