Coverage for extras/querysets.py: 84%
57 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 18:35 +0000
1from django.contrib.postgres.aggregates import JSONBAgg
2from django.db.models import Case, JSONField, OuterRef, Q, Subquery, When
4from extras.models.tags import TaggedItem
5from utilities.query_functions import EmptyGroupByJSONBAgg
6from utilities.querysets import RestrictedQuerySet
8__all__ = (
9 'ConfigContextModelQuerySet',
10 'ConfigContextQuerySet',
11 'NotificationQuerySet',
12 'SharedObjectQuerySet',
13)
16class ConfigContextQuerySet(RestrictedQuerySet):
18 def get_for_object(self, obj, aggregate_data=False):
19 """
20 Return all applicable ConfigContexts for a given object. Only active ConfigContexts will be included.
22 WARNING: This method's scope-matching logic is mirrored (inverted) by ConfigContext.get_affected_objects(),
23 which powers cache invalidation. Any change to the matching criteria here MUST be applied there as well, or
24 pre-rendered config context caches will go stale. See extras/models/configs.py.
26 Args:
27 aggregate_data: If True, use the JSONBAgg aggregate function to return only the list of JSON data objects
28 """
30 # Device type and location assignment are relevant only for Devices
31 device_type = getattr(obj, 'device_type', None)
32 location = getattr(obj, 'location', None)
33 locations = location.get_ancestors(include_self=True) if location else []
35 # Get assigned cluster, group, and type (if any)
36 cluster = getattr(obj, 'cluster', None)
37 cluster_type = getattr(cluster, 'type', None)
38 cluster_group = getattr(cluster, 'group', None)
40 # Get the group of the assigned tenant, if any
41 tenant_group = obj.tenant.group if obj.tenant else None
43 # Match against the directly assigned region as well as any parent regions.
44 region = getattr(obj.site, 'region', None)
45 regions = region.get_ancestors(include_self=True) if region else []
47 # Match against the directly assigned site group as well as any parent site groups.
48 sitegroup = getattr(obj.site, 'group', None)
49 sitegroups = sitegroup.get_ancestors(include_self=True) if sitegroup else []
51 # Match against the directly assigned role as well as any parent roles.
52 device_roles = obj.role.get_ancestors(include_self=True) if obj.role else []
54 # Match against the directly assigned platform as well as any parent platforms.
55 platform = getattr(obj, 'platform', None)
56 platforms = platform.get_ancestors(include_self=True) if platform else []
58 queryset = self.filter(
59 Q(regions__in=regions) | Q(regions=None),
60 Q(site_groups__in=sitegroups) | Q(site_groups=None),
61 Q(sites=obj.site) | Q(sites=None),
62 Q(locations__in=locations) | Q(locations=None),
63 Q(device_types=device_type) | Q(device_types=None),
64 Q(roles__in=device_roles) | Q(roles=None),
65 Q(platforms__in=platforms) | Q(platforms=None),
66 Q(cluster_types=cluster_type) | Q(cluster_types=None),
67 Q(cluster_groups=cluster_group) | Q(cluster_groups=None),
68 Q(clusters=cluster) | Q(clusters=None),
69 Q(tenant_groups=tenant_group) | Q(tenant_groups=None),
70 Q(tenants=obj.tenant) | Q(tenants=None),
71 Q(tags__slug__in=obj.tags.slugs()) | Q(tags=None),
72 is_active=True,
73 ).order_by('weight', 'name').distinct()
75 if aggregate_data: 75 ↛ 80line 75 didn't jump to line 80 because the condition on line 75 was always true
76 return queryset.aggregate(
77 config_context_data=JSONBAgg('data', order_by=['weight', 'name'])
78 )['config_context_data']
80 return queryset
83class ConfigContextModelQuerySet(RestrictedQuerySet):
84 """
85 QuerySet manager used by models which support ConfigContext (device and virtual machine).
87 Includes a method which appends an annotation of aggregated config context JSON data objects. This is
88 implemented as a subquery which performs all the joins necessary to filter relevant config context objects.
89 This offers a substantial performance gain over ConfigContextQuerySet.get_for_object() when dealing with
90 multiple objects. This allows the annotation to be entirely optional.
91 """
92 def annotate_config_context_data(self, only_invalidated=False):
93 """
94 Attach the subquery annotation to the base queryset.
96 Args:
97 only_invalidated: If True, evaluate the (expensive) aggregation subquery only for rows
98 whose pre-rendered cache (`_config_context_data`) is NULL, returning NULL for rows
99 that already have a populated cache. This is the list/detail read-path optimization:
100 warm rows are served from the cache by ConfigContextModel.get_config_context() and
101 never consult this annotation, so computing it for them is wasted work. PostgreSQL
102 short-circuits CASE branches, so the correlated SubPlan is not executed for warm
103 rows.
105 NOTE: With only_invalidated=True the annotation is NULL for warm rows. It is only
106 safe to read via get_config_context() (which short-circuits on the cache before
107 touching the annotation). Do NOT call render_config_context() directly on a row
108 annotated this way, or a warm row would render an empty context.
109 """
110 from extras.models import ConfigContext
111 subquery = Subquery(
112 ConfigContext.objects.filter(
113 self._get_config_context_filters()
114 ).annotate(
115 _data=EmptyGroupByJSONBAgg('data', order_by=['weight', 'name'])
116 ).values("_data").order_by()
117 )
118 if only_invalidated: 118 ↛ 124line 118 didn't jump to line 124 because the condition on line 118 was always true
119 subquery = Case(
120 When(_config_context_data__isnull=True, then=subquery),
121 default=None,
122 output_field=JSONField(),
123 )
124 return self.annotate(config_context_data=subquery)
126 def _get_config_context_filters(self):
127 # Construct the set of Q objects for the specific object types
128 tag_query_filters = {
129 "object_id": OuterRef(OuterRef('pk')),
130 "content_type__app_label": self.model._meta.app_label,
131 "content_type__model": self.model._meta.model_name
132 }
133 base_query = Q(
134 Q(cluster_types=OuterRef('cluster__type')) | Q(cluster_types=None),
135 Q(cluster_groups=OuterRef('cluster__group')) | Q(cluster_groups=None),
136 Q(clusters=OuterRef('cluster')) | Q(clusters=None),
137 Q(tenant_groups=OuterRef('tenant__group')) | Q(tenant_groups=None),
138 Q(tenants=OuterRef('tenant')) | Q(tenants=None),
139 Q(sites=OuterRef('site')) | Q(sites=None),
140 Q(
141 tags__pk__in=Subquery(
142 TaggedItem.objects.filter(
143 **tag_query_filters
144 ).values_list(
145 'tag_id',
146 flat=True
147 ).distinct()
148 )
149 ) | Q(tags=None),
150 is_active=True,
151 )
153 # Apply Location & DeviceType filters only for VirtualMachines
154 if self.model._meta.model_name == 'device':
155 base_query.add(
156 (Q(
157 locations__path__ancestor_or_equal=OuterRef('location__path'),
158 ) | Q(locations=None)),
159 Q.AND
160 )
161 base_query.add((Q(device_types=OuterRef('device_type')) | Q(device_types=None)), Q.AND)
162 elif self.model._meta.model_name == 'virtualmachine': 162 ↛ 168line 162 didn't jump to line 168 because the condition on line 162 was always true
163 base_query.add(Q(locations=None), Q.AND)
164 base_query.add(Q(device_types=None), Q.AND)
166 # Ltree-based filters: the ConfigContext-side tree node must be an ancestor
167 # (or equal to) the device/VM-side tree node, i.e. `cc_node.path @> obj_node.path`.
168 base_query.add(
169 (Q(
170 regions__path__ancestor_or_equal=OuterRef('site__region__path'),
171 ) | Q(regions=None)),
172 Q.AND
173 )
174 base_query.add(
175 (Q(
176 site_groups__path__ancestor_or_equal=OuterRef('site__group__path'),
177 ) | Q(site_groups=None)),
178 Q.AND
179 )
180 base_query.add(
181 (Q(
182 roles__path__ancestor_or_equal=OuterRef('role__path'),
183 ) | Q(roles=None)),
184 Q.AND
185 )
186 base_query.add(
187 (Q(
188 platforms__path__ancestor_or_equal=OuterRef('platform__path'),
189 ) | Q(platforms=None)),
190 Q.AND
191 )
193 return base_query
196class NotificationQuerySet(RestrictedQuerySet):
198 def unread(self):
199 """
200 Return only unread notifications.
201 """
202 return self.filter(read__isnull=True)
205class SharedObjectQuerySet(RestrictedQuerySet):
207 def restrict_to_shared(self, user):
208 """
209 Restrict the queryset to objects which are shared or owned by the given user. Superusers are exempt;
210 anonymous users see only shared objects. This enforces consistent visibility across the UI, REST API,
211 and GraphQL API.
212 """
213 if user.is_superuser: 213 ↛ 215line 213 didn't jump to line 215 because the condition on line 213 was always true
214 return self
215 if user.is_anonymous:
216 return self.filter(shared=True)
217 return self.filter(
218 Q(shared=True) | Q(user=user)
219 )