Django-celery-beat загружает задание, но не выполняет?

У меня есть приложение, структурированное следующим образом.

- api
  -api
    settings.py
    celery.py
  -core
    tasks.py
    -scripts
      cgm.py

При выполнении следующей команды я вижу, что моя задача загружается в базу данных, однако на самом деле она не выполняется, и я пытаюсь понять почему.

celery -A api beat -l debug -S django_celery_beat.schedulers.DatabaseScheduler

Вот мой код.

settings.py (соответствующие части)

INSTALLED_APPS = (
    ...,
    'django_celery_beat',
)


CELERY_BEAT_SCHEDULER = 'django_celery_beat.schedulers:DatabaseScheduler'

celery.py

import os

from celery import Celery

os.environ.setdefault('DJANGO_SETTINGS_MODULE',
                      'api.settings')

app = Celery('api')

app.config_from_object('django.conf:settings', namespace='CELERY')
app.autodiscover_tasks()

tasks.py

from django_celery_beat.models import PeriodicTask, IntervalSchedule


schedule, created = IntervalSchedule.objects.get_or_create(
    every=10,
    period=IntervalSchedule.SECONDS,
)


PeriodicTask.objects.get_or_create(
    interval=schedule,
    name='Import Dexcom Data',
    task='core.scripts.cgm.load_dexcom',
)

cgm.py

from monitor.models import GlucoseMonitor


def load_dexcom():
    from core.models import User
    user = User.objects.get(username='xxx')
    from pydexcom import Dexcom
    dexcom = Dexcom("xxx", "xxx", ous=True)  # add ous=True if outside of US
    bg = dexcom.get_current_glucose_reading()
    data = GlucoseMonitor.objects.create(
        user = user,
        source = 1,
        blood_glucose = bg.mmol_l,
        trend = bg.trend,
        created = bg.time
    )
    data.save()

Я могу запустить load_dexcom() вручную, и он работает. Я предполагаю, что я не правильно выполняю задачу, и она не находит ее, но она не показывает никаких ошибок в коде. Когда я запускаю команду celery, я вижу, что она загружает запись, но не делает ничего другого.

Я не уверен, что это точно поможет, так как я комбинирую некоторые кусочки из различных вещей, которые у меня работают здесь, но думаю, если вы измените tasks.py на что-то вроде следующего, это может помочь. Я думаю, что вам нужно украсить функцию, которую вы запускаете, но я не использую Django для этого, поэтому не уверен на 100%.

from django_celery_beat.models import PeriodicTask, IntervalSchedule
from core.scripts.cgm import load_dexcom


schedule, created = IntervalSchedule.objects.get_or_create(
    every=10,
    period=IntervalSchedule.SECONDS,
)


PeriodicTask.objects.get_or_create(
    interval=schedule,
    name='Import Dexcom Data',
    task=run_load_dexcom(),
)

@app.task
def run_load_dexcom():
    load_dexcom()  # and doing the appropriate import at top of file
Вернуться на верх