Handling circular import with blueprints & async tasks using rq

Question

I am trying to use rq & redis for handling asynchronous tasks. Requirement is that the task will do some operations in background and will update the database directly.

So far, I am able to do the required setup and tasks work perfectly as long as no database interaction is done. When I incorporated the code to trigger database query, I got an error that the application context is needed. Now, when I try to use that, I started getting error for circular import. I am already using blueprints but could not figure out the way to fix it.

I started with this boilerplate and then modified/added new files as below :-

app/utils/worker.py

import os
import redis
from rq import Worker, Queue, Connection

listen = ['high', 'low', 'default']

redis_url = os.getenv('REDIS_URL', 'redis://localhost:6379/0')

conn = redis.from_url(redis_url)

if __name__ == '__main__':
    with Connection(conn):
        worker = Worker(map(Queue, listen))
        worker.work()

app/utils/tasks.py

import sys
import time
from flask import Blueprint

from app.auth.models import User
from app.config import Config
from app import db, create_app

task_bp = Blueprint('task_bp', __name__)

flask_app = create_app(Config)
flask_app.app_context().push()

def background_task():
    print("Task Started")
    user = User.query.get(1)
    print(user)
    time.sleep(10)
    print("Task Completed")

app/__init__.py

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_bcrypt import Bcrypt
from flask_login import LoginManager
from flask_mail import Mail
from flask_migrate import Migrate
from app.config import Config

from app.utils.worker import conn
from rq import Queue
import rq_dashboard

db = SQLAlchemy()
migrate = Migrate()
bcrypt = Bcrypt()
login_manager = LoginManager()
login_manager.login_view = 'auth_bp.login'
login_manager.login_message_category = 'info'
mail = Mail()

q = Queue(connection=conn)

def create_app(config_class=Config):
    flask_app = Flask(__name__)
    flask_app.config.from_object(Config)

    db.init_app(flask_app)
    migrate.init_app(flask_app,db)
    bcrypt.init_app(flask_app)
    login_manager.init_app(flask_app)
    mail.init_app(flask_app)

    from app.auth.routes import auth_bp
    from app.main.routes import main_bp
    from app.errors.handlers import errors_bp
    from app.utils.tasks import task_bp
    flask_app.register_blueprint(auth_bp)
    flask_app.register_blueprint(main_bp)
    flask_app.register_blueprint(errors_bp)
    flask_app.register_blueprint(task_bp)

    flask_app.config.from_object(rq_dashboard.default_settings)
    flask_app.register_blueprint(rq_dashboard.blueprint, url_prefix="/rq")

    return flask_app

app/main/routes.py

from flask import render_template, request, Blueprint
from app import db, q

from app.utils.tasks import background_task

main_bp = Blueprint('main_bp', __name__)

@main_bp.route("/")
@main_bp.route("/home")
def home():
    return render_template('home.html', title='Home')

@main_bp.route("/about")
def about():
    q.enqueue(background_task)
    return render_template('about.html', title='About')

Error :-

  File "<frozen importlib._bootstrap>", line 219, in _call_with_frames_removed
  File "./app/utils/tasks.py", line 11, in <module>
    flask_app = create_app(Config)
  File "./app/__init__.py", line 34, in create_app
    from app.main.routes import main_bp
  File "./app/main/routes.py", line 4, in <module>
    from app.utils.tasks import background_task
ImportError: cannot import name 'background_task' from 'app.utils.tasks' (./app/utils/tasks.py)

Appreciate if someone can help me resolve it and suggest some best practices using rq and blueprints.

Answer 1

I was able to resolve it by moving the import statements inside the task function body. I am not sure if this is the right way to resolve it or if there is other clean solution for the same.

import sys
import time
from app.auth.models import User

def background_task():
    from app.config import Config
    from app import db, create_app
    flask_app = create_app(Config)
    flask_app.app_context().push()
    print("Task Started")
    user = User.query.get(1)
    print(user)
    time.sleep(10)
    print("Task Completed")