flaskredisrq

Flask and rq-scheduler: Working outside of application context


I want to start a background job that does a periodic sync when I start the flask application. This works well, but in the task function I want to access a logger that was defined outside. While this also works in the responsible class, the application context does not seem to be available in the task itself.

init.py (folder app)

from flask import Flask
from redis import Redis
from rq import Queue
from rq_scheduler import Scheduler
from datetime import datetime
import logging
from app.data.sync import Sync


def create_app():

    app = Flask(__name__)

    logging.basicConfig(filename="tasks.log", level=logging.INFO, format=f'%(asctime)s %(levelname)s %(name)s %(threadName)s : %(message)s')
    app.logger.info("Start Service")

    # Within the blueprints app can be accessed.
    #
    # from app.auth import bp as auth_bp
    # app.register_blueprint(auth_bp, url_prefix='/auth')

    with app.app_context():
        redis = Redis()
        queue = Queue('tasks', connection=redis)
        scheduler = Scheduler(queue=queue, connection=redis)

        sync = Sync()
        sync.logging_test()

        scheduler.schedule(scheduled_time=datetime.utcnow(),
                           func=sync.data_sync,
                           interval=10,
                           repeat=3
                           )

    return app

sync.py (folder app)

from flask import current_app
from time import sleep


class Sync():

    def logging_test(self):
        current_app.logger.info("this is running fine")

    def data_sync(self):
        #current_app.app_context().push()

        current_app.logger.info("Start sync")
        print("task")
        sleep(2)
        print("another task")
        current_app.logger.info("End sync")

app.py (outside)

from app import create_app


create_app()

I got the following error:

Traceback (most recent call last):
  File "/home/user/project/venv/lib/python3.10/site-packages/rq/worker.py", line 1068, in perform_job
    rv = job.perform()
  File "/home/user/project/venv/lib/python3.10/site-packages/rq/job.py", line 847, in perform
    self._result = self._execute()
  File "/home/user/project/venv/lib/python3.10/site-packages/rq/job.py", line 870, in _execute
    result = self.func(*self.args, **self.kwargs)
  File "/home/user/project/./app/data/sync.py", line 14, in data_sync
    current_app.logger.info("Start sync")
  File "/home/user/project/venv/lib/python3.10/site-packages/werkzeug/local.py", line 316, in __get__
    obj = instance._get_current_object()  # type: ignore[misc]
  File "/home/user/project/venv/lib/python3.10/site-packages/werkzeug/local.py", line 513, in _get_current_object
    raise RuntimeError(unbound_message) from None
RuntimeError: Working outside of application context.

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

Solution

  • I've found that in rq functions, there really doesn't seem to be any other way to get the Flask Context other than to recreate the app.

        app = create_app()
        app.app_context().push()
    

    This creates a new app, but it has access to the config.

    The only downside is that you have to be careful to really only create general things in create_app(). If you start a job here, it will be started every time the function is called. Since this also happens if you use several workers, for example with gunicorn, a job handling must be implemented here.