在 Flask 应用程序中访问应用程序配置变量

Question

我有一个如下所示的烧瓶应用：

myapp
-- application.py
-- models.py
-- queries.py
-- routes.py
-- settings.py

application.py 看起来像这样：

from flask import Flask
from myapp import routes

def create_app(config_object):

    app = Flask(__name__)
    app.config.from_object(config_object)

    app.register_blueprint(routes.main)

    return app

queries.py 看起来像这样

from myapp.models import User

class QueryConnection(DATABASE_URI):
    self.engine = create_engine(DATABASE_URI)
    self.session = sessionmaker(bind=self.engine)
    self.Session = self.session()

    def get_all_users(self):
        return self.Session.query(User).all()

routes.py 看起来像这样：

from flask import current_app, Blueprint, render_template
from myapp import queries
main = Blueprint('main', __name__, url_prefix='/')

query_connection = queries.QueryConnection(current_app.config['DATABASE_URI'])

@main.route("/")
def index():
    return render_template('index.jinja2', list=query_connection.get_all_users())

models.py 看起来像这样：

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'

    id           = Column(Integer, primary_key=True)
    name         = Column(String)

我想在 routes.py 中获取 current_app.config['DATABASE_URI'] 以便查询数据库，但我不知道如何将应用程序上下文从 application.py 传递给 queries.py。我尝试在 application.py 中使用它

with app.app_context():
    app.register_blueprint(main)

但我收到“在应用程序上下文之外工作”错误。其他任何东西都会给出同样的错误。

我想将查询封装在它自己的类中，这样我就可以在不同的上下文中传递数据库 uri，比如在测试中。

Answer 1

您需要从调用 create_app 方法的位置导入文件。我猜是application.py。

--- D_file.py

app = create_app(your_config)

--- queries.py

from D_file import app

DB_URI = app.config['DATABASE_URI']

Answer 2

为什么在 SQLAlchemy 设置中涉及到 Flask 应用程序？你有一个可用的配置对象，直接使用它。然后您可以将会话附加到烧瓶全局对象，或动态创建会话。

在querys.py中

import get_config from config

config_object = get_config()
engine = create_engine(config_object['DATABASE_URI'])
session_factory = sessionmaker(bind=engine)
Session = scoped_session(session_factory)

def get_all_users():
    session = Session()
    users = session.query(User).all()
    session.close()
    return users

在您的应用工厂中：

from flask import Flask, g
from myapp import routes
from myapp.queries import Session

def create_app(config_object):

    app = Flask(__name__)
    app.config.from_object(config_object)

    app.register_blueprint(routes.main)

    # automatically set up and close a session for each request
    @app.before_request
    def before_request(response):
        g.session = Session()

    @app.after_request
    def after_request(response):
        g.session.close()

    return app