Source code for actstream.actions

import datetime

from django.utils.translation import ugettext_lazy as _
from django.utils.six import text_type
from django.contrib.contenttypes.models import ContentType

from actstream import settings
from actstream.signals import action
from actstream.registry import check
from actstream.compat import get_model

    from django.utils import timezone
    now =
except ImportError:
    now =

[docs]def follow(user, obj, send_action=True, actor_only=True, **kwargs): """ Creates a relationship allowing the object's activities to appear in the user's stream. Returns the created ``Follow`` instance. If ``send_action`` is ``True`` (the default) then a ``<user> started following <object>`` action signal is sent. Extra keyword arguments are passed to the action.send call. If ``actor_only`` is ``True`` (the default) then only actions where the object is the actor will appear in the user's activity stream. Set to ``False`` to also include actions where this object is the action_object or the target. Example:: follow(request.user, group, actor_only=False) """ check(obj) instance, created = get_model('actstream', 'follow').objects.get_or_create( user=user,, content_type=ContentType.objects.get_for_model(obj), actor_only=actor_only) if send_action and created: action.send(user, verb=_('started following'), target=obj, **kwargs) return instance
[docs]def unfollow(user, obj, send_action=False): """ Removes a "follow" relationship. Set ``send_action`` to ``True`` (``False is default) to also send a ``<user> stopped following <object>`` action signal. Example:: unfollow(request.user, other_user) """ check(obj) get_model('actstream', 'follow').objects.filter( user=user,, content_type=ContentType.objects.get_for_model(obj) ).delete() if send_action: action.send(user, verb=_('stopped following'), target=obj)
[docs]def is_following(user, obj): """ Checks if a "follow" relationship exists. Returns True if exists, False otherwise. Example:: is_following(request.user, group) """ check(obj) return get_model('actstream', 'follow').objects.filter( user=user,, content_type=ContentType.objects.get_for_model(obj) ).exists()
[docs]def action_handler(verb, **kwargs): """ Handler function to create Action instance upon action signal call. """ kwargs.pop('signal', None) actor = kwargs.pop('sender') # We must store the unstranslated string # If verb is an ugettext_lazyed string, fetch the original string if hasattr(verb, '_proxy____args'): verb = verb._proxy____args[0] newaction = get_model('actstream', 'action')( actor_content_type=ContentType.objects.get_for_model(actor),, verb=text_type(verb), public=bool(kwargs.pop('public', True)), description=kwargs.pop('description', None), timestamp=kwargs.pop('timestamp', now()) ) for opt in ('target', 'action_object'): obj = kwargs.pop(opt, None) if obj is not None: check(obj) setattr(newaction, '%s_object_id' % opt, setattr(newaction, '%s_content_type' % opt, ContentType.objects.get_for_model(obj)) if settings.USE_JSONFIELD and len(kwargs): = kwargs return newaction