使用蓝图烧瓶时出现外部上下文错误 python

Outside context error when working from blueprint flask python

我用 python (Flask)

编写了这个简单的网络应用程序

models.py

from flask_sqlalchemy import SQLAlchemy
db = SQLAlchemy()

class Coin(db.Model):
    __tablename__ = "coins"
    id = db.Column(db.Integer, primary_key=True)
    pair = db.Column(db.String)
    sell_amt = db.Column(db.Float)
    buy_amt = db.Column(db.Float)

app.py

from flask import Flask 
from ui import ui 
from models import db , Coin 

app = Flask(__name__)
app.register_blueprint(ui)
db.init_app(app)

if __name__ == "__main__":
    app.run(port=8080)

__init__.py 在 ui 文件夹中

from flask import Blueprint ,current_app
from models import db, Coin
from threading import Thread
ui = Blueprint('ui', __name__)

def intro():
    global bot_state
    with current_app.app_context():
        all_coins = Coin.query.filter_by().all()
    while bot_state:
        sleep(3)
        print (f" Current time : {time()}")


@ui.route('/startbot')
def start_bot():
    global bot_thread, bot_state
    bot_state = True
    
    bot_thread = Thread(target=intro ,daemon=True)
    bot_thread.start()
    return "bot started " 

@ui.route('/stopbot')
def stop_bot():
    global bot_state 
    bot_state = False
    bot_thread.join()
    return  " bot stopped"

当创建对 /startbot 的请求时,应用程序抛出错误,因为它在应用程序上下文之外工作

RuntimeError: Working outside of application context.

This typically means that you attempted to use functionality that needed
to interface with the current application object in some way. To solve
this, set up an application context with app.app_context().  See the
documentation for more information.

但是当尝试创建一个数据库对象时,例如 new = Coin() 它工作正常,你如何在不创建 returns 应用程序的函数的情况下为函数提供应用程序的上下文,因为这样做因此会产生另一个错误(循环导入)

请注意,这是最低限度的示例,还有其他文件需要ui重新访问 models.py 文件夹(以将订单添加到机器人创建的数据库)

诀窍是将应用程序对象传递给线程。这也适用于代理 current_app。然而,在这种情况下,您需要访问底层应用程序对象。您可以在文档 here.

中找到关于此的简短说明
from flask import current_app

# ... 

def intro(app):
    with app.app_context():
        all_coins = Coin.query.all()

@ui.route('/startbot')
def start_bot():
    bot_thread = Thread(
        target=intro,
        args=(current_app._get_current_object(),), # <- !!!
        daemon=True
    )
    bot_thread.start()
    return "bot started"

由于您似乎没有完全理解我的解释,以下是 __init__.py 文件的完整内容。

from flask import Blueprint, current_app, render_template
from models import Coin, db
from threading import Event, Lock, Thread
from time import sleep, time

ui = Blueprint('ui', __name__)
thread = None
thread_event = Event()
thread_lock = Lock()

def intro(app, event):
    app.logger.info('bot started')
    try:
        while event.is_set():
            tm = time()
            app.logger.info('current time %s', tm)

            with app.app_context():
                all_coins = Coin.query.all()
                # ...

            dur = 3 - (time() - tm)
            if dur > 0: sleep(dur)
    finally:
        event.clear()
        app.logger.info('bot stopped')

@ui.route('/startbot')
def start_bot():
    global thread
    thread_event.set()
    with thread_lock:
        if thread is None:
            thread = Thread(
                target=intro,
                args=(current_app._get_current_object(), thread_event),
                daemon=True
            )
            thread.start()
    return '', 200

@ui.route('/stopbot')
def stop_bot():
    global thread
    thread_event.clear()
    with thread_lock:
        if thread is not None:
            thread.join()
            thread = None
    return '', 200

在您的项目的进一步实施中获得乐趣和成功。

必须有更好的方法,但这是我设法做到的,我们创建了两个应用程序,第一个是主要的网络应用程序,看起来像这样

app = Flask(__name__)
app.register_blueprint(some_blueprint)
db.init_app(app)

第二个应用程序将用于机器人,将在编写机器人核心代码的同一文件中声明,可以导入到蓝图中,如下所示

bot_app = Flask(__name__)
db.init_app(app)

现在intro会变成这样

from bot_file import bot_app

def intro(app):
    with bot_app.app_context():
        all_coins = Coin.query.all()

这样我们就可以在 bot_core class 中使用 bot_app 而无需导入主网络应用程序

这不是最好的代码,但它确实解决了这个问题