# settings.py # CELERY STUFF BROKER_URL = 'redis://localhost:6379' CELERY_RESULT_BACKEND = 'redis://localhost:6379' CELERY_ACCEPT_CONTENT = ['application/json'] CELERY_TASK_SERIALIZER = 'json' CELERY_RESULT_SERIALIZER = 'json'
# celery.py from __future__ import absolute_import import os from celery import Celery from django.conf import settings # os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'backend.settings') app = Celery('backend') # app.config_from_object('django.conf:settings') app.autodiscover_tasks(lambda: settings.INSTALLED_APPS) @app.task(bind=True) def debug_task(self): print('Request: {0!r}'.format(self.request))
# tasks.py import json import requests from celery.decorators import task from celery.task.schedules import crontab from celery.decorators import periodic_task from celery.utils.log import get_task_logger from api.controllers.message import message_delayed logger = get_task_logger(__name__) @periodic_task( run_every=(crontab(minute='*/1')), ignore_result=True ) def period(): # req = requests.get('') req = message_delayed() send(req) logger.info('task success') @task def send(mes): print('it works') # do_something
структура
в view.py оборачиваю функцию в task, добавил вывод print() чтоб посмотреть в worker работает или нет, в итоге ничего не выполняется, на серваке метод дергается, в task.py все работает отлично
При запуске worker выводит инфу что она находит этот таск
[tasks] . api.tasks.period . api.tasks.send . api.views.user_get . backend.celery.debug_task
Все по идеи должно работать, но в итоге я не знаю работает ли, кто что подскажет?