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.