Flask and rq-scheduler: Working outside of application context

Viewed 26

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.
0 Answers
Related