Господа, но как в целери запилить периодически вызываемый таск? всю

Question

Господа, но как в целери запилить периодически вызываемый таск? всю

голову сломал. есть метод у обьекта сельдерея add_periodic_task(), в него пихается расписание и таск, который по нему вызывается, как я понял.
В доке указано просто add_periodic_task(schedule, sig, args=(), kwargs=(), name=None, **opts)
Вот мой таск для теста:
@celery.task
def test(data):
celery_logger.info(f'It works and {data}')
Вот таск для старта теста из роута фласка:
@celery.task(name='tasks.start')
def test_task(hour=7):
celery_logger.info('It works, bitches')
celery.add_periodic_task(crontab(hour=hour),
test.s('hello'), name='asdasd')
при запуска старта вылетает ошибка:
"test() missing 1 required positional argument: 'data'"

#flask #programming #russian

0

05.01.2021

2 ответов

12 просмотров

Gleb B Автор вопроса

Alexander
Возможно там первый аргумент ждет экземпляр класса...

ога, спс, попробую

0

05.01.2021

Alexander · Accepted Answer

Alexander

Возможно там первый аргумент ждет экземпляр класса. А у тебя видимо вне класс юзается. Попробуй передать просто что-нибудь раз он его ждет, напирмер test.s(True, ‘hello')

0

05.01.2021

169 похожих чатов

Господа, но как в целери запилить периодически вызываемый таск? всю

2 ответов

Похожие вопросы