Django信号基于datetime字段值

Posted 2023-02-01

我正在努力解决以下问题。我正在尝试创建一个自定义信号，当当前时间等于我模型的notify_on DateTimeField值时触发。

像这样的东西：

class Notification(models.Model):
    ...
    notify_on = models.DateTimeField()



def send_email(*args, **kwargs):
    # send email


signals.when_its_time.connect(send_email, sender=User)

在我阅读完所有文档之后，我没有找到关于如何实现这种信号的信息。

有任何想法吗？

更新：不太天真的方法，能够丢弃不相关的任务：https://stackoverflow.com/a/55337663/9631956

答案

在django的文档中，有两个有趣的信号可以帮助你完成这项任务：pre_save和post_save。这取决于您的需求，但是假设您想要在保存模型后检查模型的notify_on是否等于当前日期（实际上在调用save()或create()方法之后）。如果是你的情况，你可以这样做：

from datetime import datetime
from django.contrib.auth.models import User
from django.db import models
from django.dispatch import receiver
from django.db.models.signals import post_save


class Notification(models.Model):
    ...
    # Every notification is related to a user
    # It depends on your model, but i guess you're doing something similar
    user = models.ForeignKey(User, related_name='notify', on_delete=models.DO_NOTHING)
    notify_on = models.DateTimeField()
    ...
    def send_email(self, *args, **kwargs):
        """A model method to send email notification"""
        ...


@receiver(post_save, sender=User)
def create_notification(sender, instance, created, **kwargs):
    # check if the user instance is created
    if created:
        obj = Notification.objects.create(user=instance, date=datetime.now().date())
        if obj.notify_on == datetime.now().date():
            obj.send_email()

你应该知道，只有当有一个触发它们的动作时，django信号才会被它们拥有。这意味着Django信号不会遍历模型的实例并执行操作，但当应用程序对连接到信号的模型执行操作时，会触发django信号。

额外奖励：要对您的实例执行循环并定期处理操作，您可能需要一个asyncworker与Queue数据库（主要是Celery与Redis or RabbitMQ）。

另一答案

好的，感谢@SergeyPugach的评论我做了以下事情：

添加了一个post_save信号，该信号调用一个向芹菜添加任务的功能。 apply_async让你通过eta - 预计到达时间可以直接接受DateTimeField，非常方便。

# models.py
from django.db.models import signals
from django.db import models
from .tasks import send_notification

class Notification(models.Model):
    ...
    notify_on = models.DateTimeField()


def notification_post_save(instance, *args, **kwargs):
    send_notification.apply_async((instance,), eta=instance.notify_on)


signals.post_save.connect(notification_post_save, sender=Notification)

和tasks.py的实际任务

import logging
from user_api.celery import app
from django.core.mail import send_mail
from django.template.loader import render_to_string


@app.task
def send_notification(self, instance):
    try:
        mail_subject = 'Your notification.'
        message = render_to_string('notify.html', 
            'title': instance.title,
            'content': instance.content
        )
        send_mail(mail_subject, message, recipient_list=[instance.user.email], from_email=None)
    except instance.DoesNotExist:
        logging.warning("Notification does not exist anymore")

我不会详细介绍如何设置芹菜，那里有很多信息。

现在我将尝试弄清楚在更新通知实例后如何更新任务，但这是一个完全不同的故事。