【发布时间】:2019-07-19 01:42:57
【问题描述】:
我想从我保存的对象中获取absolute url。我的模型有一个名为 get_absolute_url 的方法,我想用我的 django post_save signal 调用这个方法。
当在名为Thread 的特定表中添加新条目时,我收到post_save signal。这个 post_save 信号执行我的 Celery 任务。
我的线程模型是:
class Thread(models.Model):
""" A thread with a title """
topic = models.ForeignKey('Topic')
title = models.CharField(max_length=200)
sticky = models.BooleanField(default=False)
slug = models.SlugField()
time_created = models.DateTimeField(default=timezone.now)
time_last_activity = models.DateTimeField(default=timezone.now)
def __init__(self, *args, **kwargs):
""" Initialize 'time_last_activity' to 'time_created' """
super(Thread, self).__init__(*args, **kwargs)
self.time_last_activity = self.time_created
def __str__(self):
""" Return the thread's title """
return self.title
def get_absolute_url(self):
""" Return the url of the instance's detail view """
url_kwargs = {
'topic_pk': self.topic.pk,
'topic_slug': self.topic.slug,
'thread_pk': self.pk,
'thread_slug': self.slug,
}
return reverse('simple-forums:thread-detail', kwargs=url_kwargs)
在我的模型中,我有一个 celery.py 文件:
# -*- coding: utf-8 -*-
from __future__ import unicode_literals
from django.db.models.signals import post_save
from django.dispatch import receiver
from simple_forums.models import Thread
from ..tasks import thread_notification
@receiver(post_save, sender=Thread)
def get_new_thread(sender, instance, **kwargs):
""" Post_save method which start Celery task to notify forum subscribers that a new thread has been created """
url = Thread.get_absolute_url()
print(url)
thread_title = instance.title
thread_id = instance.id
topic_id = instance.topic_id
topic_slug = instance.topic.slug
topic_title = instance.topic.title
thread_notification.delay(thread_id=thread_id, thread_title=thread_title, topic_id=topic_id, topic_slug=topic_slug,
topic_title=topic_title)
在我的 tasks.py 文件中:
# -*- coding: utf-8 -*-
from celery import shared_task
from django.contrib.auth import get_user_model
from django.utils.translation import ugettext_lazy as _
User = get_user_model()
@shared_task(bind=True, time_limit=3600, soft_time_limit=3600)
def thread_notification(self):
print('Celery task executed')
return ['success_message', _('Celery task ended')]
我想获取 absolute_url 以便使用新的 Thread 路径发送电子邮件。
我的问题是:如果我没有特定的视图(没有必要),我如何获取get_absolute_url 或使用request.build_absolute_uri?
【问题讨论】: