Outside context error when working from blueprint flask python-CodePudding

I have this simple webapp written in 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 in ui folder

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"

When create a request to /startbot the app throws the error the it is working outside the app context

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.

but when trying to create a database object for example new = Coin() it works fine, how do you give a function the context of the application without making a function that returns the app, because doing so creates another error that is (circular import)

Note this is the bare minimum example and there are other files that require access to the models.py folder (to add orders to the data base created by the bot )

CodePudding user response：

There has to be a better way of doing it but this is what I managed to do, we create two apps the first one is the main web app, and looks sth like this

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

and the second app will be for the bot and will be declared in the same file where the bot core code is written and can be imported into the blueprint and looks like this

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

Now intro will look sth like this

from bot_file import bot_app

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

this way we can use the bot_app in the bot_core class with out importing the main web app

This isn't the most preferable code out there but it does solve this problem

CodePudding user response：

The trick is to pass the application object to the thread. This also works with the proxy current_app. In this case, however, you need access to the underlying application object. You can find a short note on this within the documentation 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"

Since you don't seem to have fully understood my explanations, the following is how the complete contents of the __init__.py file would look like.

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

Have fun and success with the further implementation of your project.