django - 在 django 中使用 Celery 执行周期性任务的问题
问题描述
我的芹菜定期任务不起作用。我希望根据日期每晚更新我的数据库。这是我在应用程序目录中的 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__'
'''
解决方案
解决方案 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 "beat" 来触发周期性任务:
celery -A app worker -B
解决方案 2
但是 - 对于所描述的情况,老实说,我不明白为什么要绕道在一个不同的数据库字段中添加信息,无论如何每个记录都可以轻松地得出这些信息。这增加了数据库本身的冗余,似乎不需要。
为什么不直接使用 aManager
来轻松获取所需的模型实例?
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()
推荐阅读
- angular - 单击提交按钮后,角度将输入设置为空白
- javascript - 如何修复我的 React Context 以呈现 useEffect 并重新加载表数据
- django - 简单的 django 频道应用程序没有与 daphne 一起运行
- apache-beam - 使用 BEAM 2.27 + FLINK 1.10 的 docker 模式将 Minio S3 凭证传递给 python SDK Harness
- websocket - Delphi 10.4:Indy TIdTCPClient 从 websocket 和 webserver 读取数据
- pandas - 如何使用 for 循环更有效地编写 pandas 命令
- python-3.x - 组合 2 个数字时缺少情节标题
- php - 发布请求在 while 循环中不起作用
- c# - 使用 C# System.Text.Json 对带有混合大小写的属性名称的 json 文本进行序列化/反序列化
- python - 如何在文本文档中找到所有单词