Skip to content
Find file
Fetching contributors…
Cannot retrieve contributors at this time
167 lines (149 sloc) 6.52 KB
"""Definitions of Celery tasks in Askbot
in this module there are two types of functions:
* those wrapped with a @task decorator and a ``_celery_task`` suffix - celery tasks
* those with the same base name, but without the decorator and the name suffix
the actual work units run by the task
Celery tasks are special functions in a way that they require all the parameters
be serializable - so instead of ORM objects we pass object id's and
instead of query sets - lists of ORM object id's.
That is the reason for having two types of methods here:
* the base methods (those without the decorator and the
``_celery_task`` in the end of the name
are work units that are called from the celery tasks.
* celery tasks - shells that reconstitute the necessary ORM
objects and call the base methods
import sys
import traceback
from django.contrib.contenttypes.models import ContentType
from django.template import Context
from django.utils.translation import ugettext as _
from celery.decorators import task
from askbot.conf import settings as askbot_settings
from askbot import const
from askbot import mail
from askbot.models import Post, Thread, User, ReplyAddress
from askbot.models.badges import award_badges_signal
# TODO: Make exceptions raised inside record_post_update_celery_task() ...
# ... propagate upwards to test runner, if only CELERY_ALWAYS_EAGER = True
# (i.e. if Celery tasks are not deferred but executed straight away)
@task(ignore_result = True)
def notify_author_of_published_revision_celery_task(revision):
#todo: move this to ``askbot.mail`` module
#for answerable email only for now, because
#we don't yet have the template for the read-only notification
if askbot_settings.REPLY_BY_EMAIL:
#generate two reply codes (one for edit and one for addition)
#to format an answerable email or not answerable email
reply_options = {
'reply_action': 'append_content'
append_content_address = ReplyAddress.objects.create_new(
reply_options['reply_action'] = 'replace_content'
replace_content_address = ReplyAddress.objects.create_new(
#populate template context variables
reply_code = append_content_address + ',' + replace_content_address
if == 'question':
mailto_link_subject =
mailto_link_subject = _('An edit for my answer')
#todo: possibly add more mailto thread headers to organize messages
prompt = _('To add to your post EDIT ABOVE THIS LINE')
reply_separator_line = const.SIMPLE_REPLY_SEPARATOR_TEMPLATE % prompt
data = {
'site_name': askbot_settings.APP_SHORT_NAME,
'replace_content_address': replace_content_address,
'reply_separator_line': reply_separator_line,
'mailto_link_subject': mailto_link_subject,
'reply_code': reply_code
#load the template
from askbot.skins.loaders import get_template
template = get_template('email/notify_author_about_approved_post.html')
#todo: possibly add headers to organize messages in threads
headers = {'Reply-To': append_content_address}
#send the message
subject_line = _('Your post at %(site_name)s is now published') % data,
body_text = template.render(Context(data)),
recipient_list = [,],
related_object = revision,
activity_type = const.TYPE_ACTIVITY_EMAIL_UPDATE_SENT,
headers = headers
@task(ignore_result = True)
def record_post_update_celery_task(
newly_mentioned_user_id_list = None,
updated_by_id = None,
timestamp = None,
created = False,
diff = None,
#reconstitute objects from the database
updated_by = User.objects.get(id=updated_by_id)
post_content_type = ContentType.objects.get(id=post_content_type_id)
post = post_content_type.get_object_for_this_type(id=post_id)
newly_mentioned_users = User.objects.filter(
notify_sets = post.get_notify_sets(
#todo: take into account created == True case
#update_object is not used
(activity_type, update_object) = post.get_updated_activity_data(created)
except Exception:
# HACK: exceptions from Celery job don't propagate upwards
# to the Django test runner
# so at least let's print tracebacks
print >>sys.stderr, traceback.format_exc()
@task(ignore_result = True)
def record_question_visit(
question_post = None,
user = None,
update_view_count = False):
"""celery task which records question visit by a person
updates view counter, if necessary,
and awards the badges associated with the
question visit
#1) maybe update the view count
#question_post = Post.objects.filter(
# id = question_post_id
if update_view_count:
if user.is_anonymous():
#2) question view count per user and clear response displays
#user = User.objects.get(id = user_id)
if user.is_authenticated():
#get response notifications
#3) send award badges signal for any badges
#that are awarded for question views
event = 'view_question',
actor = user,
context_object = question_post,
Jump to Line
Something went wrong with that request. Please try again.