CODEWITHSUNDEEP
pythonflaskapscheduler
Andrew Gill
Andrew Gill

Reputation: 361

How to run an apscheduler job which requires app_context

I have a flask app which needs to run some of the methods as background tasks. I have been trying to use apscheduler. Background tasks which do not require app_context run without issue, however, tasks which do require it always throw an error:

RuntimeError: Working outside of application context.

I have tried various options. 1. I have passed app into the job, and altered all the jobs to accept app as a parameter.

I have tried to force the background task to start an app with the following:

class APScheduler(_BaseAPScheduler):
    def start(self):
        app = create_app()
        apply_config(app)
        with app.app_context():
            super().start()

Both options do not appear to have managed to get app_context. Are there any other ways to force the background task to have app_context?

Upvotes: 4

Views: 2505

Answers (2)

jpmulongo
jpmulongo

Reputation: 31

Try using the Flask-APScheduler extension. It's a convenient wrapper around the apscheduler library.

So for your use case, you would do something like this

from flask_apscheduler import APScheduler

scheduler = APScheduler()

# Then to use a flask context inside a job
def some_job():
    with scheduler.app.app_context():
      #...

Upvotes: 2

Seto Kaiba
Seto Kaiba

Reputation: 140

You might have already figured it out, but here is my approach. I would like to you what was yours

In global context create the app

#Let BackgroundJobs be A module having all logics of batch jobs
from BackgroundJobs import BatchJobs 
app = Flask(__name__)

def job1():
     with app.app_context():
          BatchJobs.job1()

def job2():
     with app.app_context():
          BatchJobs.job2()



scheduler = BackgroundScheduler()
scheduler.add_job(job1, 'interval', minutes=5)
scheduler.add_job(job2, 'interval', minutes=5)

scheduler.start()
if __name__ == '__main__':
    app.run(host='0.0.0.0', debug=False)

Inside BackgroundJobs.py, you will have to import proxy to app using below. To have app available, we should invoke the methods in module with app.app_conext()

from flask import current_app as app

def job1():
     app.logger.info('Job1')

def job2():
     app.logger.info('Job2')

Upvotes: 0

Related Questions