Celery

Что есть что и как запускать?

celeryd

celeryd - celery-worker - выполняет таски, прибывающие на celery

Запуск:

python manage.py celeryd -Q celery,backup,hl7,duplicates --loglevel=INFO --purge

celerybeat

celerybeat - celery-планировщик - добавляет в очередь (брокер сообщений) задачи по расписанию (напр. через крон)

Запуск:

python manage.py celerybeat --loglevel=INFO

Как отменять таски?

# lock.core.LockService.unlock
from www.celery_conf import app

app.control.revoke(task_id, terminate=True)

Источник

Внутри таски

# hl7_server.tasks.client_query

@shared_task(bind=True)
def client_query(self, service_hl7_id, message_er7, task_id=None):
    self.app.control.revoke(task_id, terminate=True)

Важно: нужно указывать bind=True, чтобы в таске был self

Как посмотреть отмененные таски?

app.control.inspect().revoked()

Как запускать cron по таймзоне?

Сделать partial-функцию, у которой nowfun будет генерить дату с таймзоной:

import datetime as dt
from functools import partial
import pytz
from celery.schedules import crontab
from django.conf import settings

tz_crontab = partial(
    crontab,
    nowfun=lambda: dt.datetime.now(pytz.timezone(settings.TIME_ZONE)),
)

CELERYBEAT_SCHEDULE = {
    'load_daily_stats': {
        'task': 'statistics.tasks.load_daily_stats',
        'schedule': tz_crontab(hour=0, minute=0),
    },
}