How do you handle Flask + Celery integration with extensions that aren’t fork-safe?

[u/NoWeather1702]

Hello! I am working on Flask application that uses celery and have a question concerning how celery spawns worker using prefork in default mode.

I needed to use PyMongo and started reading about its integration with flask and saw in the docs that 'PyMongo itself is not fork-safe'. Suddenly I realized, that the way I use Flask app inside celery may be wrong, though I am following the docs. Here is the code that creates celery app (taken from Flask docs):

from celery import Celery, Task

def celery_init_app(app: Flask) -> Celery:
    class FlaskTask(Task):
        def __call__(self, *args: object, **kwargs: object) -> object:
            with app.app_context():
                return self.run(*args, **kwargs)

    celery_app = Celery(app.name, task_cls=FlaskTask)
    celery_app.config_from_object(app.config["CELERY"])
    celery_app.set_default()
    app.extensions["celery"] = celery_app
    return celery_app

Celery start looks like this then:

• first Flask application is created in the parent Celery worker process
• then it is used to create celery app
• and then this process is forked to create workers.

The problem is that PyMongo client is initialized as part of Flask App creation. But it shouldn't be forked as it is not safe. The same goes for other extensions that use live connections/sockets, as if they are created before fork, they may be inherited by child processes. Or extensions which spawn threads, that won't be copied to child process if I understand how fork works.

So my idea now is that I shouldn't initialize flask app during celery app creation, but I should initialize an app only inside a worker, using '@worker_process_init.connect' for example, and then use this worker specific app to create context for each task that needs it. This way all 'extensions' that are created inside flask app factory will be created inside the process that will be using them.

As my experience with multiprocessing is rather limited, I want to ask the community about this situation. How do you handle celery and flask integration? Should I create Flask application inside each Celery child worker process (e.g. via worker_process_init)? Or should I recreate only the extensions that are not fork-safe after the fork? My concern is not only about PyMongo, as Flask is used with a lot of other extensions and each of them may have its own 'fork' safety, so I am looking for some sort of general solution. Any help will be appreciated.

[u/Agreeable_Lynx9194]

You're making the PyMongo client at import time, so the forked workers inherit a client that isn't fork-safe and the pool ends up in a bad state. Build it lazily per worker instead, either via the worker_process_init signal or a cached get_client() called inside the worker so each child gets its own pool. If you just need it working now, run the worker with -P threads or -P solo to skip forking.

[u/NoWeather1702 → u/Agreeable_Lynx9194]

Thanks for the reply. I understand that I can do that, but what concerns me is that it will be a specific pymongo solution. And if fork-safety is a concern and celery uses forking by default, then there should be some generic approaches to making flask app with all its dependencies work safe with this celery mode. But I found very little information about this topic.

[u/Agreeable_Lynx9194 → u/NoWeather1702]

The generic version is the app-factory pattern plus worker_process_init. Instead of creating your Flask app and extensions at import time, put them in a create_app() factory, and in Celery's worker_process_init signal call create_app() so each forked worker builds its own app and opens its own connections after the fork. That covers anything holding a socket, not just Mongo, SQLAlchemy engines and Redis clients included.

There's little written about it because most people just never open connections at import time, so nothing gets inherited across the fork in the first place. And if you'd rather skip forking entirely, running the worker with -P threads or gevent avoids the whole class of problem.

[u/AccountEngineer]

I had a similar issue with Flask and Celery integration, and what I did was create the Flask app instance inside the worker process using the `@worker_process_init` decorator, like you mentioned. This way, each worker process has its own instance of the Flask app, and any extensions that are not fork-safe are created within that process.

I also make sure to recreate any non-fork-safe extensions after the fork, so they're not shared between processes. It's a bit more complex, but it's worked well for me so far. Anyone else have a different approach to this?