7009540190

Committed 27 Nov 2023 06:58PM UTC coverage: 94.394% (-0.6%) from 95.009%

Build # 7009540190

Build Type

Pull #534

github

Committed by

web-flow

Commit Message

Merge c0f354781 into a7df85e3a

Pull Request Pull Request #534: Integrated swappable model support

Run Details

238 of 302 branches covered (0.0%)

99 of 113 new or added lines in 18 files covered. (87.61%)

2 existing lines in 1 file now uncovered.

1549 of 1641 relevant lines covered (94.39%)

11.33 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

92.75

/actstream/feeds.py

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_str
from django.views.generic import View
from django.http import HttpResponse, Http404
from django.urls import reverse

from actstream.models import model_stream, user_stream, any_stream
from actstream.settings import get_action_model


class AbstractActivityStream:
    """
    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 = date.strftime('%Y-%m-%d')
        return 'tag:{},{}:{}'.format(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': str(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': str(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_str(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:

    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:

    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 StreamKwargsMixin:

    def items(self, request, *args, **kwargs):
        return self.get_stream()(
            self.get_object(request, *args, **kwargs),
            **self.get_stream_kwargs(request)
        )



class UserActivityMixin:

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

    def get_stream(self):
        return user_stream

    def get_stream_kwargs(self, request):
        stream_kwargs = {}
        if 'with_user_activity' in request.GET:
            stream_kwargs['with_user_activity'] = request.GET['with_user_activity'].lower() == 'true'
        return stream_kwargs


class CustomStreamMixin:
    name = None

    def get_object(self):
        return

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

justquick / django-activity-stream / 7009540190

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous