FollowManager.following_qs() - Code Metrics - justquick/django-activity-stream - Measure and Improve Code Quality continuously with Scrutinizer

FollowManager.following_qs() A
last analyzed 2018-06-29 09:47 UTC

↳ Parent: FollowManager

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	3
Bugs	0	Features	0

Metric	Value
cc	3
c	3
b	0
f	0
dl	0
loc	18
rs	9.5

from django.apps import apps
from django.db.models import Subquery, Q
from django.contrib.contenttypes.models import ContentType

from actstream.gfk import GFKManager
from actstream.decorators import stream
from actstream.registry import check


class ActionManager(GFKManager):
    """
    Default manager for Actions, accessed through Action.objects
    """

    def public(self, *args, **kwargs):
        """
        Only return public actions
        """
        kwargs['public'] = True
        return self.filter(*args, **kwargs)

    @stream
    def actor(self, obj, **kwargs):
        """
        Stream of most recent actions where obj is the actor.
        Keyword arguments will be passed to Action.objects.filter
        """
        check(obj)
        return obj.actor_actions.public(**kwargs)

    @stream
    def target(self, obj, **kwargs):
        """
        Stream of most recent actions where obj is the target.
        Keyword arguments will be passed to Action.objects.filter
        """
        check(obj)
        return obj.target_actions.public(**kwargs)

    @stream
    def action_object(self, obj, **kwargs):
        """
        Stream of most recent actions where obj is the action_object.
        Keyword arguments will be passed to Action.objects.filter
        """
        check(obj)
        return obj.action_object_actions.public(**kwargs)

    @stream
    def model_actions(self, model, **kwargs):
        """
        Stream of most recent actions by any particular model
        """
        check(model)
        ctype = ContentType.objects.get_for_model(model)
        return self.public(
            (Q(target_content_type=ctype) |
             Q(action_object_content_type=ctype) |
             Q(actor_content_type=ctype)),
            **kwargs
        )

    @stream
    def any(self, obj, **kwargs):
        """
        Stream of most recent actions where obj is the actor OR target OR action_object.
        """
        check(obj)
        ctype = ContentType.objects.get_for_model(obj)
        return self.public(
            Q(
                actor_content_type=ctype,
                actor_object_id=obj.pk,
            ) | Q(
                target_content_type=ctype,
                target_object_id=obj.pk,
            ) | Q(
                action_object_content_type=ctype,
                action_object_object_id=obj.pk,
            ), **kwargs)

    @stream
    def user(self, obj, with_user_activity=False, **kwargs):
        """Create a stream of the most recent actions by objects that the user is following."""
        q = Q()
        qs = self.public()

        if not obj:
            return qs.none()

        check(obj)

        if with_user_activity:
            q = q | Q(
                actor_content_type=ContentType.objects.get_for_model(obj),
                actor_object_id=obj.pk
            )

        follows = apps.get_model('actstream', 'follow').objects.filter(user=obj)
        content_types = ContentType.objects.filter(
            pk__in=Subquery(follows.values('content_type_id'))
        )

        if not (content_types.exists() or with_user_activity):
            return qs.none()

        for content_type in content_types:
            object_ids = follows.filter(content_type=content_type)
            q = q | Q(
                actor_content_type=content_type,
                actor_object_id__in=Subquery(object_ids.values('object_id'))
            ) | Q(
                target_content_type=content_type,
                target_object_id__in=Subquery(
                    object_ids.filter(actor_only=False).values('object_id')
                )
            ) | Q(
                action_object_content_type=content_type,
                action_object_object_id__in=Subquery(
                    object_ids.filter(actor_only=False).values('object_id')
                )
            )

        return qs.filter(q, **kwargs)


class FollowManager(GFKManager):
    """
    Manager for Follow model.
    """

    def for_object(self, instance, flag=''):
        """
        Filter to a specific instance.
        """
        check(instance)
        content_type = ContentType.objects.get_for_model(instance).pk
        queryset = self.filter(content_type=content_type, object_id=instance.pk)
        if flag:
            queryset = queryset.filter(flag=flag)
        return queryset

    def is_following(self, user, instance, flag=''):
        """
        Check if a user is following an instance.
        """
        if not user or user.is_anonymous:
            return False
        queryset = self.for_object(instance)

        if flag:
            queryset = queryset.filter(flag=flag)
        return queryset.filter(user=user).exists()

    def followers_qs(self, actor, flag=''):
        """
        Returns a queryset of User objects who are following the given actor (eg my followers).
        """
        check(actor)
        queryset = self.filter(
            content_type=ContentType.objects.get_for_model(actor),
            object_id=actor.pk
        ).select_related('user')

        if flag:
            queryset = queryset.filter(flag=flag)
        return queryset

    def followers(self, actor, flag=''):
        """
        Returns a list of User objects who are following the given actor (eg my followers).
        """
        return [follow.user for follow in self.followers_qs(actor, flag=flag)]

    def following_qs(self, user, *models, **kwargs):
        """
        Returns a queryset of actors that the given user is following (eg who im following).
        Items in the list can be of any model unless a list of restricted models are passed.
        Eg following(user, User) will only return users following the given user
        """
        qs = self.filter(user=user)
        ctype_filters = Q()
        for model in models:
            check(model)
            ctype_filters |= Q(content_type=ContentType.objects.get_for_model(model))
        qs = qs.filter(ctype_filters)

        flag = kwargs.get('flag', '')

        if flag:
            qs = qs.filter(flag=flag)
        return qs.fetch_generic_relations('follow_object')

    def following(self, user, *models, **kwargs):
        """
        Returns a list of actors that the given user is following (eg who im following).
        Items in the list can be of any model unless a list of restricted models are passed.
        Eg following(user, User) will only return users following the given user
        """
        return [follow.follow_object for follow in self.following_qs(
            user, *models, flag=kwargs.get('flag', '')
        )]


1			from django.apps import apps
2			from django.db.models import Subquery, Q
3			from django.contrib.contenttypes.models import ContentType
4
5			from actstream.gfk import GFKManager
6			from actstream.decorators import stream
7			from actstream.registry import check
8
9
10			class ActionManager(GFKManager):
11			"""
12			Default manager for Actions, accessed through Action.objects
13			"""
14
15			def public(self, args, *kwargs):
16			"""
17			Only return public actions
18			"""
19			kwargs['public'] = True
20			return self.filter(args, *kwargs)
21
22			@stream
23			def actor(self, obj, **kwargs):
24			"""
25			Stream of most recent actions where obj is the actor.
26			Keyword arguments will be passed to Action.objects.filter
27			"""
28			check(obj)
29			return obj.actor_actions.public(**kwargs)
30
31			@stream
32			def target(self, obj, **kwargs):
33			"""
34			Stream of most recent actions where obj is the target.
35			Keyword arguments will be passed to Action.objects.filter
36			"""
37			check(obj)
38			return obj.target_actions.public(**kwargs)
39
40			@stream
41			def action_object(self, obj, **kwargs):
42			"""
43			Stream of most recent actions where obj is the action_object.
44			Keyword arguments will be passed to Action.objects.filter
45			"""
46			check(obj)
47			return obj.action_object_actions.public(**kwargs)
48
49			@stream
50			def model_actions(self, model, **kwargs):
51			"""
52			Stream of most recent actions by any particular model
53			"""
54			check(model)
55			ctype = ContentType.objects.get_for_model(model)
56			return self.public(
57			(Q(target_content_type=ctype) \|
58			Q(action_object_content_type=ctype) \|
59			Q(actor_content_type=ctype)),
60			**kwargs
61			)
62
63			@stream
64			def any(self, obj, **kwargs):
65			"""
66			Stream of most recent actions where obj is the actor OR target OR action_object.
67			"""
68			check(obj)
69			ctype = ContentType.objects.get_for_model(obj)
70			return self.public(
71			Q(
72			actor_content_type=ctype,
73			actor_object_id=obj.pk,
74			) \| Q(
75			target_content_type=ctype,
76			target_object_id=obj.pk,
77			) \| Q(
78			action_object_content_type=ctype,
79			action_object_object_id=obj.pk,
80			), **kwargs)
81
82			@stream
83			def user(self, obj, with_user_activity=False, **kwargs):
84			"""Create a stream of the most recent actions by objects that the user is following."""
85			q = Q()
86			qs = self.public()
87
88			if not obj:
89			return qs.none()
90
91			check(obj)
92
93			if with_user_activity:
94			q = q \| Q(
95			actor_content_type=ContentType.objects.get_for_model(obj),
96			actor_object_id=obj.pk
97			)
98
99			follows = apps.get_model('actstream', 'follow').objects.filter(user=obj)
100			content_types = ContentType.objects.filter(
101			pk__in=Subquery(follows.values('content_type_id'))
102			)
103
104			if not (content_types.exists() or with_user_activity):
105			return qs.none()
106
107			for content_type in content_types:
108			object_ids = follows.filter(content_type=content_type)
109			q = q \| Q(
110			actor_content_type=content_type,
111			actor_object_id__in=Subquery(object_ids.values('object_id'))
112			) \| Q(
113			target_content_type=content_type,
114			target_object_id__in=Subquery(
115			object_ids.filter(actor_only=False).values('object_id')
116			)
117			) \| Q(
118			action_object_content_type=content_type,
119			action_object_object_id__in=Subquery(
120			object_ids.filter(actor_only=False).values('object_id')
121			)
122			)
123
124			return qs.filter(q, **kwargs)
125
126
127			class FollowManager(GFKManager):
128			"""
129			Manager for Follow model.
130			"""
131
132			def for_object(self, instance, flag=''):
133			"""
134			Filter to a specific instance.
135			"""
136			check(instance)
137			content_type = ContentType.objects.get_for_model(instance).pk
138			queryset = self.filter(content_type=content_type, object_id=instance.pk)
139			if flag:
140			queryset = queryset.filter(flag=flag)
141			return queryset
142
143			def is_following(self, user, instance, flag=''):
144			"""
145			Check if a user is following an instance.
146			"""
147			if not user or user.is_anonymous:
148			return False
149			queryset = self.for_object(instance)
150
151			if flag:
152			queryset = queryset.filter(flag=flag)
153			return queryset.filter(user=user).exists()
154
155			def followers_qs(self, actor, flag=''):
156			"""
157			Returns a queryset of User objects who are following the given actor (eg my followers).
158			"""
159			check(actor)
160			queryset = self.filter(
161			content_type=ContentType.objects.get_for_model(actor),
162			object_id=actor.pk
163			).select_related('user')
164
165			if flag:
166			queryset = queryset.filter(flag=flag)
167			return queryset
168
169			def followers(self, actor, flag=''):
170			"""
171			Returns a list of User objects who are following the given actor (eg my followers).
172			"""
173			return [follow.user for follow in self.followers_qs(actor, flag=flag)]
174
175			def following_qs(self, user, models, *kwargs):
176			"""
177			Returns a queryset of actors that the given user is following (eg who im following).
178			Items in the list can be of any model unless a list of restricted models are passed.
179			Eg following(user, User) will only return users following the given user
180			"""
181			qs = self.filter(user=user)
182			ctype_filters = Q()
183			for model in models:
184			check(model)
185			ctype_filters \|= Q(content_type=ContentType.objects.get_for_model(model))
186			qs = qs.filter(ctype_filters)
187
188			flag = kwargs.get('flag', '')
189
190			if flag:
191			qs = qs.filter(flag=flag)
192			return qs.fetch_generic_relations('follow_object')
193
194			def following(self, user, models, *kwargs):
195			"""
196			Returns a list of actors that the given user is following (eg who im following).
197			Items in the list can be of any model unless a list of restricted models are passed.
198			Eg following(user, User) will only return users following the given user
199			"""
200			return [follow.follow_object for follow in self.following_qs(
201			user, *models, flag=kwargs.get('flag', '')
202			)]
203

justquick / django-activity-stream

FollowManager.following_qs() A last analyzed 2018-06-29 09:47 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

FollowManager.following_qs() A
last analyzed 2018-06-29 09:47 UTC