Django 模型上的 Celery 任务

Question

我正在尝试学习如何使用 celery 在我的一个模型上每天检查日期。我的一个模型有一个到期日期和一个布尔字段，表示他们的保险是否已过期。

模型很大，所以我要 post 压缩版。我想我有两个选择。运行模型方法上的 celery 任务或重写我的 tasks.py 中的函数。然后我需要用 Celery 打到运行每天检查时间表。

我的功能可以正常工作，但我直接传入模型对象，我认为这是错误的。

我在如何使用 celery.py 中的 celery beat 调度程序中的参数方面也遇到了问题。

我真的很接近让它工作，但我想我将以错误的方式执行任务。我认为在模型方法上执行任务可能是最干净的，我只是不确定如何完成它。

models.py

class CarrierCompany(models.Model):
    name = models.CharField(max_length=255, unique=True)
    insurance_expiration = models.DateTimeField(null=True)
    insurance_active = models.BooleanField()

    def insurance_expiration_check(self):
        if self.insurance_expiration > datetime.today().date():
            self.insurance_active = True
            self.save()
            print("Insurance Active")
        else:
            self.insurance_active = False
            self.save()
            print("Insurance Inactive")

tasks.py

from __future__ import absolute_import, unicode_literals
from celery.decorators import task
from datetime import datetime, date
from django.utils import timezone
from .models import CarrierCompany



@task(name="insurance_expired")
def insurance_date():
    carriers = CarrierCompany.objects.all()
    for carrier in carriers:
        date = datetime.now(timezone.utc)
        if carrier.insurance_expiration > date:
            carrier.insurance_active = True
            carrier.save()
            print("Insurance Active")
        else:
            carrier.insurance_active = False
            carrier.save()
            print("Insurance Inactive")

celery.py

from __future__ import absolute_import, unicode_literals
import os
from celery import Celery
from celery.schedules import crontab

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'config.settings.local')

app = Celery('POTRTMS')

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

app.autodiscover_tasks()


@app.task(bind=True)
def debug_task(self):
    print('Request: {0!r}'.format(self.request))


app.conf.beat_schedule = {
    'check-insurance-daily': {
        'task': 'insurance_expired',
        'schedule': crontab(hour='8')
    },
}

*** 更新了节拍时间表以反映我真正想要运行的时间。

Answer 1

我可以如何做的一个例子如下。此外，如果您在 Django 应用程序中包含时区，而不是使用传统的日期时间，您可能希望使用此处的时区库。

models.py

class CarrierCompany(models.Model):
    ...

    @property
    def is_insurance_expired(self):
        from django.utils import timezone
        if self.insurance_expiration > timezone.datetime.today():
            print("Insurance Active")
            return True
        else:
            print("Insurance Active")
            return False

tasks.py

def insurance_date():
    carriers = CarrierCompany.objects.all()
    for carrier in carriers:
        if carrier.is_insurance_expired:
            carrier.insurance_active = True
            carrier.save()
        else:
            carrier.insurance_active = False
            carrier.save()

您还可以做其他事情，例如如果它为 False 则不更新它并设为默认 False，反之亦然。您也可以在模型函数中完成所有这些操作，尽管我个人喜欢将逻辑分开一些（就像我组织我的东西的方式一样）。希望这可以帮助您摆脱困境。

Django 模型上的 Celery 任务

Celery Tasks on Django Models

django

celery

celery-task

celerybeat