Logo Questions Linux Laravel Mysql Ubuntu Git Menu
 

How do I get the application context in a Blueprint, but not in a request?

Tags:

python

flask

I am attempting to convert a collection of Flask apps to a single app with several Blueprints.

In one of my apps, I have a task that runs periodically in the background, not related to a request. It looks something like this:

import apscheduler.schedulers.background
import flask

app = flask.Flask(__name__)
app.config['DATABASE']
scheduler = apscheduler.schedulers.background.BackgroundScheduler()
scheduler.start()

def db():
    _db = flask.g.get('_db')
    if _db is None:
        _db = get_db_connection_somehow(app.config['DATABASE'])
        flask.g._db = _db
    return _db

@scheduler.scheduled_job('interval', hours=1)
def do_a_thing():
    with app.app_context():
        db().do_a_thing()

When I convert this app to a Blueprint, I lose access to the app object and I can't figure out how to create an application context when I need one. This is what I tried:

import apscheduler.schedulers.background
import flask

bp = flask.Blueprint('my_blueprint', __name__)
scheduler = apscheduler.schedulers.background.BackgroundScheduler()
scheduler.start()

def db():
    _db = flask.g.get('_db')
    if _db is None:
        _db = get_db_connection_somehow(flask.current_app.config['DATABASE'])
        flask.g._db = _db
    return _db

@bp.record
def record(state):
    with state.app.app_context():
        flask.g._app = state.app

@scheduler.scheduled_job('interval', hours=1)
def do_a_thing():
    with flask.g._app.app_context():
        db().do_a_thing()

The error I get is:

RuntimeError: Working outside of application context.

So, how can I get the application context in a blueprint but outside a request?

like image 845
William Jackson Avatar asked Aug 09 '17 20:08

William Jackson


1 Answers

I solved this problem with the following changes. First, I set up a scheduler object on my Flask app:

app = flask.Flask(__name__)
app.scheduler = apscheduler.schedulers.background.BackgroundScheduler()
app.scheduler.start()

Next, I changed the function that runs my background task to accept the app as an argument, so I could read the database connection information from app.config:

def do_a_thing(app: flask.Flask):
    db = get_db_connection_somehow(app.config['DATABASE'])
    db.do_a_thing()

Finally, I set up the scheduled job in Blueprint.record():

@bp.record
def record(state):
    state.app.scheduler.add_job(do_a_thing, trigger='interval', args=[state.app], hours=1)
like image 118
William Jackson Avatar answered Oct 18 '22 11:10

William Jackson