Updated on December 19th, 2023 in #flask

Getting Celery to Work with Python 3.12 and Flask 3.0

You may need to make a slight modification to your create_celery_app function when updating to Python 3.12.

Quick Jump: Demo Video

Over the years Flask has changed how to create a Celery app in their documentation. For example I used what’s below for many years:

def create_celery_app(app=None):
    app = app or create_app()

    celery = Celery(app.import_name)
    celery.conf.update(app.config.get("CELERY_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

However when upgrading to Python 3.12, you’ll end up getting this error. At least with Celery 5.3.6 and Kombu 5.3.4 (a dependendy of Celery):

  File "/app/hello/app.py", line 34, in create_celery_app
    celery.Task = ContextTask
    ^^^^^^^^^^^
  File "/home/python/.local/lib/python3.12/site-packages/kombu/utils/objects.py", line 37, in __set__
    with self.lock:
         ^^^^^^^^^
AttributeError: 'cached_property' object has no attribute 'lock'

I ended up reporting the issue in an open issue related to Python 3.12.

So for a while I held off on upgrading to Python 3.12. I wasn’t going to update to Python 3.12.0 anyways because for most language versions I do like waiting for a X.X.1 release unless I really want to use a new feature.

Once Python 3.12.1 dropped I started to look to see if Celery’s create app function could be modified to not depend on the above code path.

Sure enough the Flask docs had an updated version of their Celery docs and it suggested:

def create_celery_app(app=None):
    app = app or create_app()

    class FlaskTask(Task):
        def __call__(self, *args, **kwargs):
            with app.app_context():
                return self.run(*args, **kwargs)

    celery = Celery(app.import_name, task_cls=FlaskTask)
    celery.conf.update(app.config.get("CELERY_CONFIG", {}))
    celery.set_default()
    app.extensions["celery"] = celery

    return celery

It all worked, even on a large code base with dozens of Celery tasks, including recurring tasks Celery beat.

You can find a working example of this in my example Docker Flask app and I’ll be updating my Build a SAAS App with Flask course with this free update in the near future.

This was a good reminder to revisit the docs for things you’ve been using for a long time. Who knows what improvements you’ll find. This is especially important for micro frameworks like Flask where you’ll be generating a good amount of code in your apps.

Demo Video

Timestamps

• 0:24 – I ran into an error with Kombu
• 0:55 – Flask’s Celery docs are good
• 1:40 – Taking a look at the new version of how to create a Celery app

Did it work for you with Python 3.12? Let me know below!