AtomModelActivityFeed - Code Metrics - justquick/django-activity-stream - Measure and Improve Code Quality continuously with Scrutinizer

AtomModelActivityFeed A
last analyzed 2018-06-29 09:47 UTC

↳ Parent: Project

Complexity

Total Complexity

Size/Duplication

Total Lines	6
Duplicated Lines	0 %

Importance

Changes

Metric	Value
c	0
b	0
f	0
dl	0
loc	6
rs	10
wmc	0

import json

from django.shortcuts import get_object_or_404
from django.core.exceptions import ObjectDoesNotExist
from django.utils.feedgenerator import Atom1Feed, rfc3339_date
from django.contrib.contenttypes.models import ContentType
from django.contrib.syndication.views import Feed, add_domain
from django.contrib.sites.models import Site
from django.utils.encoding import force_text
from django.utils.six import text_type
from django.utils import datetime_safe
from django.views.generic import View
from django.http import HttpResponse, Http404

try:
    from django.core.urlresolvers import reverse
except ImportError:
    from django.urls import reverse

from actstream.models import Action, model_stream, user_stream, any_stream


class AbstractActivityStream(object):
    """
    Abstract base class for all stream rendering.
    Supports hooks for fetching streams and formatting actions.
    """
    def get_stream(self, *args, **kwargs):
        """
        Returns a stream method to use.
        """
        raise NotImplementedError

    def get_object(self, *args, **kwargs):
        """
        Returns the object (eg user or actor) that the stream is for.
        """
        raise NotImplementedError

    def items(self, *args, **kwargs):
        """
        Returns a queryset of Actions to use based on the stream method and object.
        """
        return self.get_stream()(self.get_object(*args, **kwargs))

    def get_uri(self, action, obj=None, date=None):
        """
        Returns an RFC3987 IRI ID for the given object, action and date.
        """
        if date is None:
            date = action.timestamp
        date = datetime_safe.new_datetime(date).strftime('%Y-%m-%d')
        return 'tag:%s,%s:%s' % (Site.objects.get_current().domain, date,
                                 self.get_url(action, obj, False))

    def get_url(self, action, obj=None, domain=True):
        """
        Returns an RFC3987 IRI for a HTML representation of the given object, action.
        If domain is true, the current site's domain will be added.
        """
        if not obj:
            url = reverse('actstream_detail', None, (action.pk,))
        elif hasattr(obj, 'get_absolute_url'):
            url = obj.get_absolute_url()
        else:
            ctype = ContentType.objects.get_for_model(obj)
            url = reverse('actstream_actor', None, (ctype.pk, obj.pk))
        if domain:
            return add_domain(Site.objects.get_current().domain, url)
        return url

    def format(self, action):
        """
        Returns a formatted dictionary for the given action.
        """
        item = {
            'id': self.get_uri(action),
            'url': self.get_url(action),
            'verb': action.verb,
            'published': rfc3339_date(action.timestamp),
            'actor': self.format_actor(action),
            'title': text_type(action),
        }
        if action.description:
            item['content'] = action.description
        if action.target:
            item['target'] = self.format_target(action)
        if action.action_object:
            item['object'] = self.format_action_object(action)
        return item

    def format_item(self, action, item_type='actor'):
        """
        Returns a formatted dictionary for an individual item based on the action and item_type.
        """
        obj = getattr(action, item_type)
        return {
            'id': self.get_uri(action, obj),
            'url': self.get_url(action, obj),
            'objectType': ContentType.objects.get_for_model(obj).name,
            'displayName': text_type(obj)
        }

    def format_actor(self, action):
        """
        Returns a formatted dictionary for the actor of the action.
        """
        return self.format_item(action)

    def format_target(self, action):
        """
        Returns a formatted dictionary for the target of the action.
        """
        return self.format_item(action, 'target')

    def format_action_object(self, action):
        """
        Returns a formatted dictionary for the action object of the action.
        """
        return self.format_item(action, 'action_object')


class ActivityStreamsAtomFeed(Atom1Feed):
    """
    Feed rendering class for the v1.0 Atom Activity Stream Spec
    """
    def root_attributes(self):
        attrs = super(ActivityStreamsAtomFeed, self).root_attributes()
        attrs['xmlns:activity'] = 'http://activitystrea.ms/spec/1.0/'
        return attrs

    def add_root_elements(self, handler):
        super(ActivityStreamsAtomFeed, self).add_root_elements(handler)

    def quick_elem(self, handler, key, value):
        if key == 'link':
            handler.addQuickElement(key, None, {
                'href': value, 'type': 'text/html', 'rel': 'alternate'})
        else:
            handler.addQuickElement(key, value)

    def item_quick_handler(self, handler, name, item):
        handler.startElement(name, {})
        for key, value in item.items():
            self.quick_elem(handler, key, value)
        handler.endElement(name)

    def add_item_elements(self, handler, item):
        item.pop('unique_id')
        actor = item.pop('actor')
        target = item.pop('target', None)
        action_object = item.pop('action_object', None)
        content = item.pop('content', None)

        if content:
            handler.addQuickElement('content', content, {'type': 'html'})

        for key, value in item.items():
            if value:
                self.quick_elem(handler, key, value)

        self.item_quick_handler(handler, 'author', actor)

        if action_object:
            self.item_quick_handler(handler, 'activity:object', action_object)

        if target:
            self.item_quick_handler(handler, 'activity:target', target)


class ActivityStreamsBaseFeed(AbstractActivityStream, Feed):

    def feed_extra_kwargs(self, obj):
        """
        Returns an extra keyword arguments dictionary that is used when
        initializing the feed generator.
        """
        return {}

    def item_extra_kwargs(self, action):
        """
        Returns an extra keyword arguments dictionary that is used with
        the `add_item` call of the feed generator.
        Add the 'content' field of the 'Entry' item, to be used by the custom
        feed generator.
        """
        item = self.format(action)
        item.pop('title', None)
        item['uri'] = item.pop('url')
        item['activity:verb'] = item.pop('verb')
        return item

    def format_item(self, action, item_type='actor'):
        name = item_type == 'actor' and 'name' or 'title'
        item = super(ActivityStreamsBaseFeed, self).format_item(action, item_type)
        item[name] = item.pop('displayName')
        item['activity:object-type'] = item.pop('objectType')
        item.pop('url')
        return item

    def item_link(self, action):
        return self.get_url(action)

    def item_description(self, action):
        if action.description:
            return force_text(action.description)

    def items(self, obj):
        return self.get_stream()(obj)[:30]


class JSONActivityFeed(AbstractActivityStream, View):
    """
    Feed that generates feeds compatible with the v1.0 JSON Activity Stream spec
    """
    def dispatch(self, request, *args, **kwargs):
        return HttpResponse(self.serialize(request, *args, **kwargs),
                            content_type='application/json')

    def serialize(self, request, *args, **kwargs):
        items = self.items(request, *args, **kwargs)
        return json.dumps({
            'totalItems': len(items),
            'items': [self.format(action) for action in items]
        }, indent=4 if 'pretty' in request.GET or 'pretty' in request.POST else None)


class ModelActivityMixin(object):

    def get_object(self, request, content_type_id):
        return get_object_or_404(ContentType, pk=content_type_id).model_class()

    def get_stream(self):
        return model_stream


class ObjectActivityMixin(object):

    def get_object(self, request, content_type_id, object_id):
        ct = get_object_or_404(ContentType, pk=content_type_id)
        try:
            obj = ct.get_object_for_this_type(pk=object_id)
        except ObjectDoesNotExist:
            raise Http404('No %s matches the given query.' % ct.model_class()._meta.object_name)
        return obj

    def get_stream(self):
        return any_stream


class UserActivityMixin(object):

    def get_object(self, request):
        if request.user.is_authenticated:
            return request.user

    def get_stream(self):
        return user_stream


class CustomStreamMixin(object):
    name = None

    def get_object(self):
        return

    def get_stream(self):
        return getattr(Action.objects, self.name)

    def items(self, *args, **kwargs):
        return self.get_stream()(*args[1:], **kwargs)


class ModelActivityFeed(ModelActivityMixin, ActivityStreamsBaseFeed):

    def title(self, model):
        return 'Activity feed from %s' % model.__name__

    def link(self, model):
        ctype = ContentType.objects.get_for_model(model)
        return reverse('actstream_model', None, (ctype.pk,))

    def description(self, model):
        return 'Public activities of %s' % model.__name__


class ObjectActivityFeed(ObjectActivityMixin, ActivityStreamsBaseFeed):
    def title(self, obj):
        return 'Activity for %s' % obj

    def link(self, obj):
        return self.get_url(obj)

    def description(self, obj):
        return 'Activity for %s' % obj


class UserActivityFeed(UserActivityMixin, ActivityStreamsBaseFeed):

    def title(self, user):
        return 'Activity feed for your followed actors'

    def link(self, user):
        if not user:
            return reverse('actstream')
        if hasattr(user, 'get_absolute_url'):
            return user.get_absolute_url()
        ctype = ContentType.objects.get_for_model(user)
        return reverse('actstream_actor', None, (ctype.pk, user.pk))

    def description(self, user):
        return 'Public activities of actors you follow'


class AtomUserActivityFeed(UserActivityFeed):
    """
    Atom feed of Activity for a given user (where actions are those that the given user follows).
    """
    feed_type = ActivityStreamsAtomFeed
    subtitle = UserActivityFeed.description


class AtomModelActivityFeed(ModelActivityFeed):
    """
    Atom feed of Activity for a given model (where actions involve the given model as any of the entities).
    """
    feed_type = ActivityStreamsAtomFeed
    subtitle = ModelActivityFeed.description


class AtomObjectActivityFeed(ObjectActivityFeed):
    """
    Atom feed of Activity for a given object (where actions involve the given object as any of the entities).
    """
    feed_type = ActivityStreamsAtomFeed
    subtitle = ObjectActivityFeed.description


class UserJSONActivityFeed(UserActivityMixin, JSONActivityFeed):
    """
    JSON feed of Activity for a given user (where actions are those that the given user follows).
    """
    pass


class ModelJSONActivityFeed(ModelActivityMixin, JSONActivityFeed):
    """
    JSON feed of Activity for a given model (where actions involve the given model as any of the entities).
    """
    pass


class ObjectJSONActivityFeed(ObjectActivityMixin, JSONActivityFeed):
    """
    JSON feed of Activity for a given object (where actions involve the given object as any of the entities).
    """
    pass


class CustomJSONActivityFeed(CustomStreamMixin, JSONActivityFeed):
    """
    JSON feed of Activity for a custom stream. self.name should be the name of the custom stream as defined in the Manager
    and arguments may be passed either in the url or when calling as_view(...)
    """
    pass


1			import json
2
3			from django.shortcuts import get_object_or_404
4			from django.core.exceptions import ObjectDoesNotExist
5			from django.utils.feedgenerator import Atom1Feed, rfc3339_date
6			from django.contrib.contenttypes.models import ContentType
7			from django.contrib.syndication.views import Feed, add_domain
8			from django.contrib.sites.models import Site
9			from django.utils.encoding import force_text
10			from django.utils.six import text_type
11			from django.utils import datetime_safe
12			from django.views.generic import View
13			from django.http import HttpResponse, Http404
14
15			try:
16			from django.core.urlresolvers import reverse
17			except ImportError:
18			from django.urls import reverse
19
20			from actstream.models import Action, model_stream, user_stream, any_stream
21
22
23			class AbstractActivityStream(object):
24			"""
25			Abstract base class for all stream rendering.
26			Supports hooks for fetching streams and formatting actions.
27			"""
28			def get_stream(self, args, *kwargs):
29			"""
30			Returns a stream method to use.
31			"""
32			raise NotImplementedError
33
34			def get_object(self, args, *kwargs):
35			"""
36			Returns the object (eg user or actor) that the stream is for.
37			"""
38			raise NotImplementedError
39
40			def items(self, args, *kwargs):
41			"""
42			Returns a queryset of Actions to use based on the stream method and object.
43			"""
44			return self.get_stream()(self.get_object(args, *kwargs))
45
46			def get_uri(self, action, obj=None, date=None):
47			"""
48			Returns an RFC3987 IRI ID for the given object, action and date.
49			"""
50			if date is None:
51			date = action.timestamp
52			date = datetime_safe.new_datetime(date).strftime('%Y-%m-%d')
53			return 'tag:%s,%s:%s' % (Site.objects.get_current().domain, date,
54			self.get_url(action, obj, False))
55
56			def get_url(self, action, obj=None, domain=True):
57			"""
58			Returns an RFC3987 IRI for a HTML representation of the given object, action.
59			If domain is true, the current site's domain will be added.
60			"""
61			if not obj:
62			url = reverse('actstream_detail', None, (action.pk,))
63			elif hasattr(obj, 'get_absolute_url'):
64			url = obj.get_absolute_url()
65			else:
66			ctype = ContentType.objects.get_for_model(obj)
67			url = reverse('actstream_actor', None, (ctype.pk, obj.pk))
68			if domain:
69			return add_domain(Site.objects.get_current().domain, url)
70			return url
71
72			def format(self, action):
73			"""
74			Returns a formatted dictionary for the given action.
75			"""
76			item = {
77			'id': self.get_uri(action),
78			'url': self.get_url(action),
79			'verb': action.verb,
80			'published': rfc3339_date(action.timestamp),
81			'actor': self.format_actor(action),
82			'title': text_type(action),
83			}
84			if action.description:
85			item['content'] = action.description
86			if action.target:
87			item['target'] = self.format_target(action)
88			if action.action_object:
89			item['object'] = self.format_action_object(action)
90			return item
91
92			def format_item(self, action, item_type='actor'):
93			"""
94			Returns a formatted dictionary for an individual item based on the action and item_type.
95			"""
96			obj = getattr(action, item_type)
97			return {
98			'id': self.get_uri(action, obj),
99			'url': self.get_url(action, obj),
100			'objectType': ContentType.objects.get_for_model(obj).name,
101			'displayName': text_type(obj)
102			}
103
104			def format_actor(self, action):
105			"""
106			Returns a formatted dictionary for the actor of the action.
107			"""
108			return self.format_item(action)
109
110			def format_target(self, action):
111			"""
112			Returns a formatted dictionary for the target of the action.
113			"""
114			return self.format_item(action, 'target')
115
116			def format_action_object(self, action):
117			"""
118			Returns a formatted dictionary for the action object of the action.
119			"""
120			return self.format_item(action, 'action_object')
121
122
123			class ActivityStreamsAtomFeed(Atom1Feed):
124			"""
125			Feed rendering class for the v1.0 Atom Activity Stream Spec
126			"""
127			def root_attributes(self):
128			attrs = super(ActivityStreamsAtomFeed, self).root_attributes()
129			attrs['xmlns:activity'] = 'http://activitystrea.ms/spec/1.0/'
130			return attrs
131
132			def add_root_elements(self, handler):
133			super(ActivityStreamsAtomFeed, self).add_root_elements(handler)
134
135			def quick_elem(self, handler, key, value):
136			if key == 'link':
137			handler.addQuickElement(key, None, {
138			'href': value, 'type': 'text/html', 'rel': 'alternate'})
139			else:
140			handler.addQuickElement(key, value)
141
142			def item_quick_handler(self, handler, name, item):
143			handler.startElement(name, {})
144			for key, value in item.items():
145			self.quick_elem(handler, key, value)
146			handler.endElement(name)
147
148			def add_item_elements(self, handler, item):
149			item.pop('unique_id')
150			actor = item.pop('actor')
151			target = item.pop('target', None)
152			action_object = item.pop('action_object', None)
153			content = item.pop('content', None)
154
155			if content:
156			handler.addQuickElement('content', content, {'type': 'html'})
157
158			for key, value in item.items():
159			if value:
160			self.quick_elem(handler, key, value)
161
162			self.item_quick_handler(handler, 'author', actor)
163
164			if action_object:
165			self.item_quick_handler(handler, 'activity:object', action_object)
166
167			if target:
168			self.item_quick_handler(handler, 'activity:target', target)
169
170
171			class ActivityStreamsBaseFeed(AbstractActivityStream, Feed):
172
173			def feed_extra_kwargs(self, obj):
174			"""
175			Returns an extra keyword arguments dictionary that is used when
176			initializing the feed generator.
177			"""
178			return {}
179
180			def item_extra_kwargs(self, action):
181			"""
182			Returns an extra keyword arguments dictionary that is used with
183			the `add_item` call of the feed generator.
184			Add the 'content' field of the 'Entry' item, to be used by the custom
185			feed generator.
186			"""
187			item = self.format(action)
188			item.pop('title', None)
189			item['uri'] = item.pop('url')
190			item['activity:verb'] = item.pop('verb')
191			return item
192
193			def format_item(self, action, item_type='actor'):
194			name = item_type == 'actor' and 'name' or 'title'
195			item = super(ActivityStreamsBaseFeed, self).format_item(action, item_type)
196			item[name] = item.pop('displayName')
197			item['activity:object-type'] = item.pop('objectType')
198			item.pop('url')
199			return item
200
201			def item_link(self, action):
202			return self.get_url(action)
203
204			def item_description(self, action):
205			if action.description:
206			return force_text(action.description)
207
208			def items(self, obj):
209			return self.get_stream()(obj)[:30]
210
211
212			class JSONActivityFeed(AbstractActivityStream, View):
213			"""
214			Feed that generates feeds compatible with the v1.0 JSON Activity Stream spec
215			"""
216			def dispatch(self, request, args, *kwargs):
217			return HttpResponse(self.serialize(request, args, *kwargs),
218			content_type='application/json')
219
220			def serialize(self, request, args, *kwargs):
221			items = self.items(request, args, *kwargs)
222			return json.dumps({
223			'totalItems': len(items),
224			'items': [self.format(action) for action in items]
225			}, indent=4 if 'pretty' in request.GET or 'pretty' in request.POST else None)
226
227
228			class ModelActivityMixin(object):
229
230			def get_object(self, request, content_type_id):
231			return get_object_or_404(ContentType, pk=content_type_id).model_class()
232
233			def get_stream(self):
234			return model_stream
235
236
237			class ObjectActivityMixin(object):
238
239			def get_object(self, request, content_type_id, object_id):
240			ct = get_object_or_404(ContentType, pk=content_type_id)
241			try:
242			obj = ct.get_object_for_this_type(pk=object_id)
243			except ObjectDoesNotExist:
244			raise Http404('No %s matches the given query.' % ct.model_class()._meta.object_name)
245			return obj
246
247			def get_stream(self):
248			return any_stream
249
250
251			class UserActivityMixin(object):
252
253			def get_object(self, request):
254			if request.user.is_authenticated:
255			return request.user
256
257			def get_stream(self):
258			return user_stream
259
260
261			class CustomStreamMixin(object):
262			name = None
263
264			def get_object(self):
265			return
266
267			def get_stream(self):
268			return getattr(Action.objects, self.name)
269
270			def items(self, args, *kwargs):
271			return self.get_stream()(args[1:], *kwargs)
272
273
274			class ModelActivityFeed(ModelActivityMixin, ActivityStreamsBaseFeed):
275
276			def title(self, model):
277			return 'Activity feed from %s' % model.__name__
278
279			def link(self, model):
280			ctype = ContentType.objects.get_for_model(model)
281			return reverse('actstream_model', None, (ctype.pk,))
282
283			def description(self, model):
284			return 'Public activities of %s' % model.__name__
285
286
287			class ObjectActivityFeed(ObjectActivityMixin, ActivityStreamsBaseFeed):
288			def title(self, obj):
289			return 'Activity for %s' % obj
290
291			def link(self, obj):
292			return self.get_url(obj)
293
294			def description(self, obj):
295			return 'Activity for %s' % obj
296
297
298			class UserActivityFeed(UserActivityMixin, ActivityStreamsBaseFeed):
299
300			def title(self, user):
301			return 'Activity feed for your followed actors'
302
303			def link(self, user):
304			if not user:
305			return reverse('actstream')
306			if hasattr(user, 'get_absolute_url'):
307			return user.get_absolute_url()
308			ctype = ContentType.objects.get_for_model(user)
309			return reverse('actstream_actor', None, (ctype.pk, user.pk))
310
311			def description(self, user):
312			return 'Public activities of actors you follow'
313
314
315			class AtomUserActivityFeed(UserActivityFeed):
316			"""
317			Atom feed of Activity for a given user (where actions are those that the given user follows).
318			"""
319			feed_type = ActivityStreamsAtomFeed
320			subtitle = UserActivityFeed.description
321
322
323			class AtomModelActivityFeed(ModelActivityFeed):
324			"""
325			Atom feed of Activity for a given model (where actions involve the given model as any of the entities).
326			"""
327			feed_type = ActivityStreamsAtomFeed
328			subtitle = ModelActivityFeed.description
329
330
331			class AtomObjectActivityFeed(ObjectActivityFeed):
332			"""
333			Atom feed of Activity for a given object (where actions involve the given object as any of the entities).
334			"""
335			feed_type = ActivityStreamsAtomFeed
336			subtitle = ObjectActivityFeed.description
337
338
339			class UserJSONActivityFeed(UserActivityMixin, JSONActivityFeed):
340			"""
341			JSON feed of Activity for a given user (where actions are those that the given user follows).
342			"""
343			pass
344
345
346			class ModelJSONActivityFeed(ModelActivityMixin, JSONActivityFeed):
347			"""
348			JSON feed of Activity for a given model (where actions involve the given model as any of the entities).
349			"""
350			pass
351
352
353			class ObjectJSONActivityFeed(ObjectActivityMixin, JSONActivityFeed):
354			"""
355			JSON feed of Activity for a given object (where actions involve the given object as any of the entities).
356			"""
357			pass
358
359
360			class CustomJSONActivityFeed(CustomStreamMixin, JSONActivityFeed):
361			"""
362			JSON feed of Activity for a custom stream. self.name should be the name of the custom stream as defined in the Manager
363			and arguments may be passed either in the url or when calling as_view(...)
364			"""
365			pass
366

justquick / django-activity-stream

AtomModelActivityFeed A last analyzed 2018-06-29 09:47 UTC

Complexity

Size/Duplication

Importance

Duplication Side-by-Side

Filter issues like

AtomModelActivityFeed A
last analyzed 2018-06-29 09:47 UTC