Using a Celery worker to interact with a SQLAlchemy DB, including knowing the user from the request

I have done plenty of research on this, including trying answers like this. It appears Celery has no access to my Flask app's context.

I know fully well my celery object, what will decorate my tasks, must have access to my Flask app's context. And I do believe it should, as I followed this guide to create my celery object. I am unsure if the confusion lies somewhere in the fact that I am using Flask-HTTPAuth.

Here is some of what I have.

def make_celery(app):
    celery = Celery(app.import_name, backend=app.config["CELERY_RESULT_BACKEND"], broker=app.config["CELERY_BROKER_URL"])
    celery.conf.update(app.config)
    TaskBase = celery.Task
    class ContextTask(TaskBase):
        abstract = True
        def __call__(self, *args, **kwargs):
            with app.app_context():
                return TaskBase.__call__(self, *args, **kwargs)
    celery.Task = ContextTask
    return celery

app = Flask(__name__)
auth = HTTPBasicAuth()
app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///flask_app.db"
app.config["CELERY_BROKER_URL"] = "redis://localhost:6379"
app.config["CELERY_RESULT_BACKEND"] = "redis://localhost:6379"
celery = make_celery(app)
db = SQLAlchemy(app)

@celery.task(bind=True, name="flask_app.item_loop")
def loop(self):
    items = g.user.items
    for item in items:
        print(item)

Running this task using Flask is a no-go, though. I try to start this function by hitting the server (while authorized!).

@app.route("/item_loop")
@auth.login_required
def item_loop():
    result = loop.delay()
    return "It's running."

But the Celery worker tells me the task raised unexpected: AttributeError("'_AppCtxGlobals' object has no attribute 'user'",), which I believe would imply, as mentioned, my celery object does not have the app context, even though I used the recommended factory pattern.

While the recoomendations in Dave and Greg's answers are valid, what they miss to highlight is the misunderstanding that you have regarding the use of an application context in a Celery task.

You have a Flask application, in which you are using Flask-HTTPAuth. You probably have a verify_password handler that sets g.user to the authenticated user. This means that while you are handling a request you can access the user as g.user. This is all good.

You also have one or more Celery workers, which are separate processes that have no direct connection to the Flask server. The only communication between the Flask server and the Celery worker processes happens over the message broker that you are using (typically Redis or RabbitMQ).

Depending on your needs, the Celery workers may need to have access to the Flask application. This is very common when using Flask extensions that store their configuration in the app.config dictionary. Two common extensions that require this are Flask-SQLAlchemy and Flask-Mail. Without access to app.config, the Celery task would have no way to open a connection to the database or to send an email, since it would not know the details of the database and/or email server.

To give the Celery workers access to the configuration, the accepted practice is to create duplicate Flask applications in each worker. These are secondary applications that are in no way connected to the actual application object used by the main Flask server. Their only purpose is to hold a copy of the original app.config dictionary that can be accessed by your task or by any Flask extensions your task is using.

So it is invalid to expect that a g.user set in the Flask server will be accessible also as g.user in the Celery task, simply because these are different g objects, from different application instances.

If you need to use the authenticated user in the Celery task, what you should do is pass the user_id (usually g.user.id) as an argument to your task. Then in your task, you can load the user from the database using this id. Hope this helps!