使用 Celery 的 django 中的定期任务问题

Question

我的芹菜周期性任务不起作用。我希望每天晚上根据日期更新我的数据库。这是我在应用程序指导中的 ptasks.py 文件：

'''
import datetime
import celery
from celery.task.schedules import crontab
from celery.decorators import periodic_task
from django.utils import timezone
@periodic_task(run_every=crontab(hour=0, minute=0))
def every_night():
tasks=Task.objects.all()
form=TaskForm()
if form.deadline<timezone.now() and form.staus=="ONGOING":
form.status="PENDING"
form.save()
'''

我在 settings.py 中使用 ampq：

'''
CELERY_BROKER_URL = 'amqp://guest:guest@localhost'
CELERYBEAT_SCHEDULER='djcelery.schedulers.DatabaseScheduler'
'''

这是我 models.py：

'''
from django.db import models
import datetime
import pytz
from django.utils import timezone
# Create your models here.
class Task(models.Model):
title=models.CharField(max_length=30)
complete=models.BooleanField(default=False)
created=models.DateTimeField(default=timezone.now())
comment=models.CharField(max_length=500, default="")
Tag1=models.CharField(max_length=10, default="Tag1")
deadline=models.DateTimeField(default=timezone.now())
status=models.CharField(max_length=15,default="ONGOING")
def __str__(self):
return self.title
'''

这是我的 forms.py：

'''
from django import forms
from django.forms import ModelForm
from .models import *
class TaskForm(forms.ModelForm):
class Meta:
model=Task
fields='__all__'
'''

Answer 1

解决方案 1

要按计划更新数据库记录，您可以执行以下操作：

<your_app>/models.py

from django.db import models
from django.utils import timezone

class Task(models.Model):
STATUS_PENDING = "pending"
STATUS_ONGOING = "ongoing"
STATUS_CHOICES = (
(STATUS_ONGOING, "ongoing"),
(STATUS_PENDING, "pending"),
)
status = models.CharField(
max_length=15, default=STATUS_ONGOING, choices=STATUS_CHOICES
)
title = models.CharField(max_length=30)
deadline = models.DateTimeField(auto_now_add=True)
complete = models.BooleanField(default=False)
def __str__(self):
return self.title

<your_app>/tasks.py

from django.utils import timezone
from celery.task.schedules import crontab
from celery.decorators import periodic_task
from .models import Task

@periodic_task(run_every=crontab(hour=0, minute=0))
def every_night():
qs = Task.objects.filter(deadline__lt=timezone.now(), status=Task.STATUS_ONGOING)
# Update whole queryset at once
qs.update(status=Task.STATUS_PENDING)
# Alternatively update one by one (e.g. if you need signals to be fired)
# for task in qs:
#     task.status = Task.STATUS_PENDING
#     task.save()

确保运行芹菜"节拍"以触发定期任务：

celery -A app worker -B 

解决方案 2

但是 - 对于所描述的情况，老实说，我不明白为什么要绕道而行，在单独的数据库字段中添加信息，无论如何都可以由每条记录轻松获得。这增加了数据库本身的冗余，似乎不需要。

为什么不直接使用Manager轻松获取所需的模型实例？

from django.db import models
from django.utils import timezone

class TaskManager(models.Manager):
def pending(self):
return (
self.get_queryset()
.filter(deadline__lt=timezone.now())
.exclude(complete=True)
)

class Task(models.Model):
title = models.CharField(max_length=30)
deadline = models.DateTimeField(auto_now_add=True)
complete = models.BooleanField(default=False)
objects = TaskManager()
def __str__(self):
return self.title

并获取"挂起"查询集：Task.objects.pending()