import datetime
from django.db.models import get_model
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
try:
from django.utils import timezone
now = timezone.now
except ImportError:
now = datetime.datetime.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, object_id=obj.pk,
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, object_id=obj.pk,
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, object_id=obj.pk,
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),
actor_object_id=actor.pk,
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, obj.pk)
setattr(newaction, '%s_content_type' % opt,
ContentType.objects.get_for_model(obj))
if settings.USE_JSONFIELD and len(kwargs):
newaction.data = kwargs
newaction.save(force_insert=True)
return newaction