17951596797

Committed 23 Sep 2025 03:46PM UTC coverage: 94.624% (-0.2%) from 94.796%

Build # 17951596797

Build Type

Pull #1432

github

Committed by

web-flow

Commit Message

Merge 98f3be8fc into 79917de8d

Pull Request Pull Request #1432: RDMO 3.0.0 🌠

Run Details

2110 of 2226 branches covered (94.79%)

22634 of 23920 relevant lines covered (94.62%)

3.78 hits per line

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

91.04

rdmo/projects/managers.py

from django.conf import settings
from django.db import models
from django.db.models import Q

from mptt.models import TreeManager
from mptt.querysets import TreeQuerySet

from rdmo.core.managers import CurrentSiteManagerMixin


class ProjectQuerySet(TreeQuerySet):

    def filter_user(self, user, filter_for_user=False):
        if user.is_authenticated:
            if user.has_perm('projects.view_project') and not filter_for_user:
                # return all projects for admins or users with model permissions (unless filter_for_user is set)
                return self.all()
            else:
                # create a filter for all project where this user is a member
                user_filter = Q(user=user)

                # create a filter for all projects which are visible for this user
                groups = user.groups.all()
                sites_filter = Q(visibility__sites=None) | Q(visibility__sites=settings.SITE_ID)
                groups_filter = Q(visibility__groups=None) | Q(visibility__groups__in=groups)
                visibility_filter = Q(visibility__isnull=False) & sites_filter & groups_filter

                # create a filter for all projects of this site (unless filter_for_user is set)
                if (user.has_perm('projects.view_project') or user.role.is_site_manager) and not filter_for_user:
                    current_site_filter = Q(site=settings.SITE_ID)
                else:
                    current_site_filter = Q()

                # create queryset by combining all three filters
                queryset = self.filter(user_filter | visibility_filter | current_site_filter)

                # add descendant projects
                for instance in queryset:
                    queryset |= instance.get_descendants()
                return queryset.distinct()
        else:
            return self.none()

    def filter_catalogs(self, catalogs=None, exclude_catalogs=None, exclude_null=True):
        catalogs_filter = Q()
        if exclude_null:
          catalogs_filter &= Q(catalog__isnull=False)
        if catalogs:
            catalogs_filter &= Q(catalog__in=catalogs)
        if exclude_catalogs:
            catalogs_filter &= ~Q(catalog__in=exclude_catalogs)
        return self.filter(catalogs_filter)

    def filter_groups(self, groups):
        if not groups:
            return self

        # need to import here to prevent circular import
        # queryset methods need to be refactored in modules otherwise
        from django.contrib.auth import get_user_model

        from rdmo.projects.models import Membership

        # users in the given groups
        users = get_user_model().objects.filter(groups__in=groups)
        # memberships for those users with role 'owner'
        memberships = Membership.objects.filter(role='owner', user__in=users)
        # projects that have those memberships
        return self.filter(memberships__in=memberships).distinct()

    def filter_projects_for_task_or_view(self, instance):
        # if View/Task is not available it should not show for any project
        if not instance.available:
            return self.none()

        # projects that have an unavailable catalog should be disregarded
        qs = self.filter(catalog__available=True)

        # when instance.catalogs is empty it applies to all
        if instance.catalogs.exists():
            qs = qs.filter(catalog__in=instance.catalogs.all())

        # when instance.sites is empty it applies to all
        if instance.sites.exists():
            qs = qs.filter(site__in=instance.sites.all())

        # when instance.groups is empty it applies to all
        if instance.groups.exists():
            qs = qs.filter_groups(instance.groups.all())

        return qs


class MembershipQuerySet(models.QuerySet):

    def filter_current_site(self):
        return self.filter(project__site=settings.SITE_ID)

    def filter_user(self, user):
        if user.is_authenticated:
            if user.has_perm('projects.view_membership'):
                return self.all()
            elif user.role.is_site_manager:
                return self.filter_current_site()
            else:
                from .models import Project
                projects = Project.objects.filter_user(user)
                return self.filter(project__in=projects)
        else:
            return self.objects.none()


class IssueQuerySet(models.QuerySet):

    def filter_current_site(self):
        return self.filter(project__site=settings.SITE_ID)

    def filter_user(self, user):
        if user.is_authenticated:
            if user.has_perm('projects.view_integration'):
                return self.all()
            elif user.role.is_site_manager:
                return self.filter_current_site()
            else:
                from .models import Project
                projects = Project.objects.filter_user(user)
                return self.filter(project__in=projects)
        else:
            return self.none()


class IntegrationQuerySet(models.QuerySet):

    def filter_current_site(self):
        return self.filter(project__site=settings.SITE_ID)

    def filter_user(self, user):
        if user.is_authenticated:
            if user.has_perm('projects.view_issue'):
                return self.all()
            elif user.role.is_site_manager:
                return self.filter_current_site()
            else:
                from .models import Project
                projects = Project.objects.filter_user(user)
                return self.filter(project__in=projects)
        else:
            return self.none()


class InviteQuerySet(models.QuerySet):

    def filter_current_site(self):
        return self.filter(project__site=settings.SITE_ID)

    def filter_user(self, user):
        if user.is_authenticated:
            if user.has_perm('projects.view_invite'):
                return self.all()
            elif user.role.is_site_manager:
                return self.filter_current_site()
            else:
                from .models import Project
                projects = Project.objects.filter(memberships__user=user, memberships__role='owner')
                return self.filter(project__in=projects)
        else:
            return self.none()


class SnapshotQuerySet(models.QuerySet):

    def filter_current_site(self):
        return self.filter(project__site=settings.SITE_ID)

    def filter_user(self, user):
        if user.is_authenticated:
            if user.has_perm('projects.view_snapshot'):
                return self.all()
            elif user.role.is_site_manager:
                return self.filter_current_site()
            else:
                from .models import Project
                projects = Project.objects.filter_user(user)
                return self.filter(project__in=projects)
        else:
            return self.none()


class ValueQuerySet(models.QuerySet):

    def filter_current_site(self):
        return self.filter(project__site=settings.SITE_ID)

    def filter_user(self, user):
        if user.is_authenticated:
            if user.has_perm('projects.view_value'):
                return self.all()
            elif user.role.is_site_manager:
                return self.filter_current_site()
            else:
                from .models import Project
                projects = Project.objects.filter_user(user)
                return self.filter(project__in=projects)
        else:
            return self.none()

    def filter_empty(self):
        return self.filter((Q(text='') | Q(text=None)) & Q(option=None) & (Q(file='') | Q(file=None)))

    def exclude_empty(self):
        return self.exclude((Q(text='') | Q(text=None)) & Q(option=None) & (Q(file='') | Q(file=None)))

    def exclude_empty_optional(self, catalog):
        optional_values = self.filter(attribute__in=[q.attribute for q in catalog.optional_questions])
        return self.exclude(id__in=optional_values.filter_empty().values_list('id', flat=True))

    def distinct_list(self):
        return self.order_by('attribute').values_list('attribute', 'set_prefix', 'set_index').distinct()

    def filter_set(self, set_value):
        # get the catalog and prefetch most elements of the catalog
        catalog = set_value.project.catalog
        catalog.prefetch_elements()

        # Get all attributes from matching elements and their descendants
        attributes = {
            descendant.attribute
            for element in (catalog.pages + catalog.questions)
            if element.attribute == set_value.attribute
            for descendant in element.descendants
        }

        # construct the set_prefix for descendants for this set
        descendants_set_prefix = \
            f'{set_value.set_prefix}|{set_value.set_index}' if set_value.set_prefix else str(set_value.set_index)

        # collect all values for this set and all descendants
        return self.filter(project=set_value.project, snapshot=set_value.snapshot, attribute__in=attributes).filter(
            Q(set_prefix=set_value.set_prefix, set_index=set_value.set_index) |
            Q(set_prefix__startswith=descendants_set_prefix)
        )


class ProjectManager(CurrentSiteManagerMixin, TreeManager):

    def get_queryset(self):
        return ProjectQuerySet(self.model, using=self._db)

    def filter_user(self, user, filter_for_user=False):
        return self.get_queryset().filter_user(user, filter_for_user)

    def filter_catalogs(self, catalogs=None, exclude_catalogs=None, exclude_null=True):
        return self.get_queryset().filter_catalogs(catalogs=catalogs, exclude_catalogs=exclude_catalogs,
                                                   exclude_null=exclude_null)
    def filter_groups(self, groups):
        return self.get_queryset().filter_groups(groups)

    def filter_projects_for_task_or_view(self, instance):
        return self.get_queryset().filter_projects_for_task_or_view(instance)

    def prefetch_ancestors(self, projects):
        # collect the constraints for all projects
        filters = Q()
        for project in projects:
            filters |= (Q(tree_id=project.tree_id) & Q(lft__lt=project.lft) & Q(rght__gt=project.rght))

        # Fetch all ancestors in one query
        ancestors = self.filter(filters).order_by('tree_id', 'lft')

        # Index by descendant id
        prefetched_ancestors = {
            project.id: [] for project in projects
        }
        for ancestor in ancestors:
            for project in projects:
                if ancestor.tree_id == project.tree_id and ancestor.lft < project.lft and ancestor.rght > project.rght:
                    prefetched_ancestors[project.id].append(ancestor)

        # add the project as last ancestor
        for project in projects:
            prefetched_ancestors[project.id].append(project)

        return prefetched_ancestors


class MembershipManager(CurrentSiteManagerMixin, models.Manager):

    def get_queryset(self):
        return MembershipQuerySet(self.model, using=self._db)

    def filter_user(self, user):
        return self.get_queryset().filter_user(user)


class IssueManager(CurrentSiteManagerMixin, models.Manager):

    def get_queryset(self):
        return IssueQuerySet(self.model, using=self._db)

    def filter_user(self, user):
        return self.get_queryset().filter_user(user)

    def active(self):
        return self.get_queryset().active()


class IntegrationManager(CurrentSiteManagerMixin, models.Manager):

    def get_queryset(self):
        return IntegrationQuerySet(self.model, using=self._db)

    def filter_user(self, user):
        return self.get_queryset().filter_user(user)


class InviteManager(CurrentSiteManagerMixin, models.Manager):

    def get_queryset(self):
        return InviteQuerySet(self.model, using=self._db)

    def filter_user(self, user):
        return self.get_queryset().filter_user(user)


class SnapshotManager(CurrentSiteManagerMixin, models.Manager):

    def get_queryset(self):
        return SnapshotQuerySet(self.model, using=self._db)

    def filter_user(self, user):
        return self.get_queryset().filter_user(user)


class ValueManager(CurrentSiteManagerMixin, models.Manager):

    def get_queryset(self):
        return ValueQuerySet(self.model, using=self._db)

    def filter_user(self, user):
        return self.get_queryset().filter_user(user)

1	from django.conf import settings	4✔
2	from django.db import models	4✔
3	from django.db.models import Q	4✔
4
5	from mptt.models import TreeManager	4✔
6	from mptt.querysets import TreeQuerySet	4✔
7
8	from rdmo.core.managers import CurrentSiteManagerMixin	4✔
9
10
11	class ProjectQuerySet(TreeQuerySet):	4✔
12
13	def filter_user(self, user, filter_for_user=False):	4✔
14	if user.is_authenticated:	4✔
15	if user.has_perm('projects.view_project') and not filter_for_user:	4✔
16	# return all projects for admins or users with model permissions (unless filter_for_user is set)
17	return self.all()	4✔
18	else:
19	# create a filter for all project where this user is a member
20	user_filter = Q(user=user)	4✔
21
22	# create a filter for all projects which are visible for this user
23	groups = user.groups.all()	4✔
24	sites_filter = Q(visibility__sites=None) \| Q(visibility__sites=settings.SITE_ID)	4✔
25	groups_filter = Q(visibility__groups=None) \| Q(visibility__groups__in=groups)	4✔
26	visibility_filter = Q(visibility__isnull=False) & sites_filter & groups_filter	4✔
27
28	# create a filter for all projects of this site (unless filter_for_user is set)
29	if (user.has_perm('projects.view_project') or user.role.is_site_manager) and not filter_for_user:	4✔
30	current_site_filter = Q(site=settings.SITE_ID)	4✔
31	else:
32	current_site_filter = Q()	4✔
33
34	# create queryset by combining all three filters
35	queryset = self.filter(user_filter \| visibility_filter \| current_site_filter)	4✔
36
37	# add descendant projects
38	for instance in queryset:	4✔
39	queryset \|= instance.get_descendants()	4✔
40	return queryset.distinct()	4✔
41	else:
42	return self.none()	4✔
43
44	def filter_catalogs(self, catalogs=None, exclude_catalogs=None, exclude_null=True):	4✔
45	catalogs_filter = Q()	×
46	if exclude_null:	×
47	catalogs_filter &= Q(catalog__isnull=False)	×
48	if catalogs:	×
49	catalogs_filter &= Q(catalog__in=catalogs)	×
50	if exclude_catalogs:	×
51	catalogs_filter &= ~Q(catalog__in=exclude_catalogs)	×
52	return self.filter(catalogs_filter)	×
53
54	def filter_groups(self, groups):	4✔
55	if not groups:	4✔
56	return self	×
57
58	# need to import here to prevent circular import
59	# queryset methods need to be refactored in modules otherwise
60	from django.contrib.auth import get_user_model	4✔
61
62	from rdmo.projects.models import Membership	4✔
63
64	# users in the given groups
65	users = get_user_model().objects.filter(groups__in=groups)	4✔
66	# memberships for those users with role 'owner'
67	memberships = Membership.objects.filter(role='owner', user__in=users)	4✔
68	# projects that have those memberships
69	return self.filter(memberships__in=memberships).distinct()	4✔
70
71	def filter_projects_for_task_or_view(self, instance):	4✔
72	# if View/Task is not available it should not show for any project
73	if not instance.available:	4✔
74	return self.none()	4✔
75
76	# projects that have an unavailable catalog should be disregarded
77	qs = self.filter(catalog__available=True)	4✔
78
79	# when instance.catalogs is empty it applies to all
80	if instance.catalogs.exists():	4✔
81	qs = qs.filter(catalog__in=instance.catalogs.all())	4✔
82
83	# when instance.sites is empty it applies to all
84	if instance.sites.exists():	4✔
85	qs = qs.filter(site__in=instance.sites.all())	4✔
86
87	# when instance.groups is empty it applies to all
88	if instance.groups.exists():	4✔
89	qs = qs.filter_groups(instance.groups.all())	4✔
90
91	return qs	4✔
92
93
94	class MembershipQuerySet(models.QuerySet):	4✔
95
96	def filter_current_site(self):	4✔
97	return self.filter(project__site=settings.SITE_ID)	4✔
98
99	def filter_user(self, user):	4✔
100	if user.is_authenticated:	4✔
101	if user.has_perm('projects.view_membership'):	4✔
102	return self.all()	4✔
103	elif user.role.is_site_manager:	4✔
104	return self.filter_current_site()	4✔
105	else:
106	from .models import Project	4✔
107	projects = Project.objects.filter_user(user)	4✔
108	return self.filter(project__in=projects)	4✔
109	else:
110	return self.objects.none()	×
111
112
113	class IssueQuerySet(models.QuerySet):	4✔
114
115	def filter_current_site(self):	4✔
116	return self.filter(project__site=settings.SITE_ID)	4✔
117
118	def filter_user(self, user):	4✔
119	if user.is_authenticated:	4✔
120	if user.has_perm('projects.view_integration'):	4✔
121	return self.all()	4✔
122	elif user.role.is_site_manager:	4✔
123	return self.filter_current_site()	4✔
124	else:
125	from .models import Project	4✔
126	projects = Project.objects.filter_user(user)	4✔
127	return self.filter(project__in=projects)	4✔
128	else:
129	return self.none()	×
130
131
132	class IntegrationQuerySet(models.QuerySet):	4✔
133
134	def filter_current_site(self):	4✔
135	return self.filter(project__site=settings.SITE_ID)	4✔
136
137	def filter_user(self, user):	4✔
138	if user.is_authenticated:	4✔
139	if user.has_perm('projects.view_issue'):	4✔
140	return self.all()	4✔
141	elif user.role.is_site_manager:	4✔
142	return self.filter_current_site()	4✔
143	else:
144	from .models import Project	4✔
145	projects = Project.objects.filter_user(user)	4✔
146	return self.filter(project__in=projects)	4✔
147	else:
148	return self.none()	×
149
150
151	class InviteQuerySet(models.QuerySet):	4✔
152
153	def filter_current_site(self):	4✔
154	return self.filter(project__site=settings.SITE_ID)	4✔
155
156	def filter_user(self, user):	4✔
157	if user.is_authenticated:	4✔
158	if user.has_perm('projects.view_invite'):	4✔
159	return self.all()	4✔
160	elif user.role.is_site_manager:	4✔
161	return self.filter_current_site()	4✔
162	else:
163	from .models import Project	4✔
164	projects = Project.objects.filter(memberships__user=user, memberships__role='owner')	4✔
165	return self.filter(project__in=projects)	4✔
166	else:
167	return self.none()	×
168
169
170	class SnapshotQuerySet(models.QuerySet):	4✔
171
172	def filter_current_site(self):	4✔
173	return self.filter(project__site=settings.SITE_ID)	4✔
174
175	def filter_user(self, user):	4✔
176	if user.is_authenticated:	4✔
177	if user.has_perm('projects.view_snapshot'):	4✔
178	return self.all()	4✔
179	elif user.role.is_site_manager:	4✔
180	return self.filter_current_site()	4✔
181	else:
182	from .models import Project	4✔
183	projects = Project.objects.filter_user(user)	4✔
184	return self.filter(project__in=projects)	4✔
185	else:
186	return self.none()	×
187
188
189	class ValueQuerySet(models.QuerySet):	4✔
190
191	def filter_current_site(self):	4✔
192	return self.filter(project__site=settings.SITE_ID)	4✔
193
194	def filter_user(self, user):	4✔
195	if user.is_authenticated:	4✔
196	if user.has_perm('projects.view_value'):	4✔
197	return self.all()	4✔
198	elif user.role.is_site_manager:	4✔
199	return self.filter_current_site()	4✔
200	else:
201	from .models import Project	4✔
202	projects = Project.objects.filter_user(user)	4✔
203	return self.filter(project__in=projects)	4✔
204	else:
205	return self.none()	×
206
207	def filter_empty(self):	4✔
208	return self.filter((Q(text='') \| Q(text=None)) & Q(option=None) & (Q(file='') \| Q(file=None)))	4✔
209
210	def exclude_empty(self):	4✔
211	return self.exclude((Q(text='') \| Q(text=None)) & Q(option=None) & (Q(file='') \| Q(file=None)))	4✔
212
213	def exclude_empty_optional(self, catalog):	4✔
214	optional_values = self.filter(attribute__in=[q.attribute for q in catalog.optional_questions])	4✔
215	return self.exclude(id__in=optional_values.filter_empty().values_list('id', flat=True))	4✔
216
217	def distinct_list(self):	4✔
218	return self.order_by('attribute').values_list('attribute', 'set_prefix', 'set_index').distinct()	4✔
219
220	def filter_set(self, set_value):	4✔
221	# get the catalog and prefetch most elements of the catalog
222	catalog = set_value.project.catalog	4✔
223	catalog.prefetch_elements()	4✔
224
225	# Get all attributes from matching elements and their descendants
226	attributes = {	4✔
227	descendant.attribute
228	for element in (catalog.pages + catalog.questions)
229	if element.attribute == set_value.attribute
230	for descendant in element.descendants
231	}
232
233	# construct the set_prefix for descendants for this set
234	descendants_set_prefix = \	4✔
235	f'{set_value.set_prefix}\|{set_value.set_index}' if set_value.set_prefix else str(set_value.set_index)
236
237	# collect all values for this set and all descendants
238	return self.filter(project=set_value.project, snapshot=set_value.snapshot, attribute__in=attributes).filter(	4✔
239	Q(set_prefix=set_value.set_prefix, set_index=set_value.set_index) \|
240	Q(set_prefix__startswith=descendants_set_prefix)
241	)
242
243
244	class ProjectManager(CurrentSiteManagerMixin, TreeManager):	4✔
245
246	def get_queryset(self):	4✔
247	return ProjectQuerySet(self.model, using=self._db)	4✔
248
249	def filter_user(self, user, filter_for_user=False):	4✔
250	return self.get_queryset().filter_user(user, filter_for_user)	4✔
251
252	def filter_catalogs(self, catalogs=None, exclude_catalogs=None, exclude_null=True):	4✔
253	return self.get_queryset().filter_catalogs(catalogs=catalogs, exclude_catalogs=exclude_catalogs,	×
254	exclude_null=exclude_null)
255	def filter_groups(self, groups):	4✔
256	return self.get_queryset().filter_groups(groups)	×
257
258	def filter_projects_for_task_or_view(self, instance):	4✔
259	return self.get_queryset().filter_projects_for_task_or_view(instance)	4✔
260
261	def prefetch_ancestors(self, projects):	4✔
262	# collect the constraints for all projects
263	filters = Q()	4✔
264	for project in projects:	4✔
265	filters \|= (Q(tree_id=project.tree_id) & Q(lft__lt=project.lft) & Q(rght__gt=project.rght))	4✔
266
267	# Fetch all ancestors in one query
268	ancestors = self.filter(filters).order_by('tree_id', 'lft')	4✔
269
270	# Index by descendant id
271	prefetched_ancestors = {	4✔
272	project.id: [] for project in projects
273	}
274	for ancestor in ancestors:	4✔
275	for project in projects:	4✔
276	if ancestor.tree_id == project.tree_id and ancestor.lft < project.lft and ancestor.rght > project.rght:	4✔
277	prefetched_ancestors[project.id].append(ancestor)	4✔
278
279	# add the project as last ancestor
280	for project in projects:	4✔
281	prefetched_ancestors[project.id].append(project)	4✔
282
283	return prefetched_ancestors	4✔
284
285
286	class MembershipManager(CurrentSiteManagerMixin, models.Manager):	4✔
287
288	def get_queryset(self):	4✔
289	return MembershipQuerySet(self.model, using=self._db)	4✔
290
291	def filter_user(self, user):	4✔
292	return self.get_queryset().filter_user(user)	4✔
293
294
295	class IssueManager(CurrentSiteManagerMixin, models.Manager):	4✔
296
297	def get_queryset(self):	4✔
298	return IssueQuerySet(self.model, using=self._db)	4✔
299
300	def filter_user(self, user):	4✔
301	return self.get_queryset().filter_user(user)	4✔
302
303	def active(self):	4✔
304	return self.get_queryset().active()	×
305
306
307	class IntegrationManager(CurrentSiteManagerMixin, models.Manager):	4✔
308
309	def get_queryset(self):	4✔
310	return IntegrationQuerySet(self.model, using=self._db)	4✔
311
312	def filter_user(self, user):	4✔
313	return self.get_queryset().filter_user(user)	4✔
314
315
316	class InviteManager(CurrentSiteManagerMixin, models.Manager):	4✔
317
318	def get_queryset(self):	4✔
319	return InviteQuerySet(self.model, using=self._db)	4✔
320
321	def filter_user(self, user):	4✔
322	return self.get_queryset().filter_user(user)	4✔
323
324
325	class SnapshotManager(CurrentSiteManagerMixin, models.Manager):	4✔
326
327	def get_queryset(self):	4✔
328	return SnapshotQuerySet(self.model, using=self._db)	4✔
329
330	def filter_user(self, user):	4✔
331	return self.get_queryset().filter_user(user)	4✔
332
333
334	class ValueManager(CurrentSiteManagerMixin, models.Manager):	4✔
335
336	def get_queryset(self):	4✔
337	return ValueQuerySet(self.model, using=self._db)	4✔
338
339	def filter_user(self, user):	4✔
340	return self.get_queryset().filter_user(user)	4✔

rdmorganiser / rdmo / 17951596797

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