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.
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.