Python script, ASCII text executable
1
import flask
2
from celery import Celery, Task
3
def initCeleryApp(app: flask.Flask):
6
class FlaskTask(Task):
7
def __call__(self, *args: object, **kwargs: object):
8
with app.app_context():
9
return self.run(*args, **kwargs)
10
celeryApp = Celery(app.name, task_cls=FlaskTask)
12
celeryApp.config_from_object(app.config["CELERY"])
13
celeryApp.set_default()
14
app.extensions["celery"] = celeryApp
15
return celeryApp
16