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?
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)
If you love us? You can donate to us via Paypal or buy me a coffee so we can maintain and grow! Thank you!
Donate Us With