@toobinks

Flask Celery Singleton как?

Хочу сделать задачу синглтоном.

utils/celery.py
def init_celery(app, celery):
    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


tasks.py
class CelerySingleton(celery.Task, Singleton):
    pass

@celery.task(base=CelerySingleton)
def calc():
# при обращении к sqlalchemy получаю
# application not registered on db instance and no application bound to current context


Я так понимаю, что в tasks.py класс CelerySingleton наследует celery.Task, который совсем не ContextTask. Но как сделать правильно?
  • Вопрос задан
  • 377 просмотров
Пригласить эксперта
Ваш ответ на вопрос

Войдите, чтобы написать ответ

Похожие вопросы
23 нояб. 2024, в 00:16
2000 руб./за проект
22 нояб. 2024, в 23:55
3000 руб./за проект
22 нояб. 2024, в 22:26
3500 руб./за проект