django-celery-beat 加载任务但不执行？

Question

我的应用程序结构如下。

- 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 命令时，我可以看到它加载了记录，但似乎没有做任何其他事情。

编辑-

看起来我错过了下面的 worker 命令运行

celery -A api worker -l DEBUG

但是输出清楚地显示找不到脚本。

The full contents of the message body was:
'[[], {}, {"callbacks": null, "errbacks": null, "chain": null,     "chord": null}]' (77b)
Traceback (most recent call last):
  File "/home/robin/miniconda3/envs/api/lib/python3.7/site-packages/celery/worker/consumer/consumer.py", line 581, in on_task_received
    strategy = strategies[type_]
KeyError: 'core.scripts.cgm.load_dexcom'

我试过以下迭代都给出密钥错误

load_dexcom
scripts.cgm.load_dexcom
api.core.scripts.cgm.load_dexcom

Answer 1

我不确定这是否会有所帮助，因为我结合了我在这里工作的各种东西的一些位，但想想如果你将 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

django-celery-beat 加载任务但不执行？

django-celery-beat loading task but not executing?

django

celery

django-celery-beat