Coverage for netbox/graphql/filter_lookups.py: 51%
220 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 18:35 +0000
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 18:35 +0000
1import re
2from enum import Enum
3from typing import Generic, TypeVar
5import strawberry
6import strawberry_django
7from django.core.exceptions import FieldDoesNotExist
8from django.db.models import Q, QuerySet
9from django.db.models.fields.related import ForeignKey, ManyToManyField, ManyToManyRel, ManyToOneRel
10from strawberry import ID
11from strawberry.directive import DirectiveValue
12from strawberry.types import Info
13from strawberry_django import (
14 ComparisonFilterLookup,
15 FilterLookup,
16 RangeLookup,
17 process_filters,
18)
20from netbox.graphql.scalars import BigInt
22# ------------------------------------------------------------------
23# JSON path validation (VM-323)
24# ------------------------------------------------------------------
26# Each segment of a JSON path may only contain alphanumerics, underscores, and
27# hyphens. Hyphens are included because JSON keys commonly use them; leading
28# underscores are permitted (e.g. _foo is a valid key name).
29_JSON_PATH_SEGMENT_RE = re.compile(r'^[A-Za-z0-9_][A-Za-z0-9_-]*$')
32def _validate_json_path(path: str) -> str:
33 """Validate a JSON traversal path for use in ORM lookups.
35 Each ``__``-separated segment must match ``[A-Za-z0-9_][A-Za-z0-9_-]*``.
36 Raises ``ValueError`` on an empty path, empty segment, or segment with
37 disallowed characters.
39 ORM operator names (``date``, ``regex``, etc.) are intentionally *not*
40 blocked here: ``JSONFilter.filter()`` always appends ``__`` to the path
41 before handing it to ``process_filters``, so a segment named ``regex``
42 becomes another level of JSON key traversal (``data__key__regex__exact``),
43 not the ORM regex transform (``data__key__regex=…``).
44 """
45 if not path:
46 raise ValueError("JSON path cannot be empty")
48 for segment in path.split('__'):
49 if not segment:
50 raise ValueError("JSON path contains consecutive or trailing '__'")
51 if not _JSON_PATH_SEGMENT_RE.match(segment):
52 raise ValueError(f"Invalid JSON path segment: {segment!r}")
54 return path
57__all__ = (
58 'ArrayLookup',
59 'BigIntegerLookup',
60 'FloatArrayLookup',
61 'FloatLookup',
62 'IntegerArrayLookup',
63 'IntegerLookup',
64 'IntegerRangeArrayLookup',
65 'JSONFilter',
66 'JSONLookup',
67 'JSONStringLookup',
68 'StringArrayLookup',
69 'TreeNodeFilter',
70)
72T = TypeVar('T')
73SKIP_MSG = 'Filter will be skipped on `null` value'
76# These JSON lookup types intentionally mirror the legacy DateFilterLookup[str],
77# TimeFilterLookup[str], and DatetimeFilterLookup[str] schema. JSON values are
78# string-backed, so the concrete strawberry-django date/time lookup classes
79# (which now ignore type parameters and warn) are deliberately not used here.
80@strawberry.input(name='StrDateFilterLookup')
81class JSONDateFilterLookup(ComparisonFilterLookup[str]):
82 year: ComparisonFilterLookup[int] | None = strawberry.UNSET
83 month: ComparisonFilterLookup[int] | None = strawberry.UNSET
84 day: ComparisonFilterLookup[int] | None = strawberry.UNSET
85 week_day: ComparisonFilterLookup[int] | None = strawberry.UNSET
86 iso_week_day: ComparisonFilterLookup[int] | None = strawberry.UNSET
87 week: ComparisonFilterLookup[int] | None = strawberry.UNSET
88 iso_year: ComparisonFilterLookup[int] | None = strawberry.UNSET
89 quarter: ComparisonFilterLookup[int] | None = strawberry.UNSET
92@strawberry.input(name='StrTimeFilterLookup')
93class JSONTimeFilterLookup(ComparisonFilterLookup[str]):
94 hour: ComparisonFilterLookup[int] | None = strawberry.UNSET
95 minute: ComparisonFilterLookup[int] | None = strawberry.UNSET
96 second: ComparisonFilterLookup[int] | None = strawberry.UNSET
97 date: ComparisonFilterLookup[int] | None = strawberry.UNSET
98 time: ComparisonFilterLookup[int] | None = strawberry.UNSET
101@strawberry.input(name='StrDatetimeFilterLookup')
102class JSONDatetimeFilterLookup(ComparisonFilterLookup[str]):
103 year: ComparisonFilterLookup[int] | None = strawberry.UNSET
104 month: ComparisonFilterLookup[int] | None = strawberry.UNSET
105 day: ComparisonFilterLookup[int] | None = strawberry.UNSET
106 week_day: ComparisonFilterLookup[int] | None = strawberry.UNSET
107 iso_week_day: ComparisonFilterLookup[int] | None = strawberry.UNSET
108 week: ComparisonFilterLookup[int] | None = strawberry.UNSET
109 iso_year: ComparisonFilterLookup[int] | None = strawberry.UNSET
110 quarter: ComparisonFilterLookup[int] | None = strawberry.UNSET
111 hour: ComparisonFilterLookup[int] | None = strawberry.UNSET
112 minute: ComparisonFilterLookup[int] | None = strawberry.UNSET
113 second: ComparisonFilterLookup[int] | None = strawberry.UNSET
114 date: ComparisonFilterLookup[int] | None = strawberry.UNSET
115 time: ComparisonFilterLookup[int] | None = strawberry.UNSET
118@strawberry.input(description='String lookups for JSON field values.')
119class JSONStringLookup:
120 """
121 String-filter type for use inside JSONLookup.
123 Equivalent to ``StrFilterLookup`` but defined explicitly so that the type
124 name remains stable and any future per-field restrictions are easy to add.
125 ``regex`` / ``i_regex`` are included: they provide no additional oracle
126 power beyond ``starts_with``, which is also present.
127 """
128 exact: str | None = strawberry_django.filter_field()
129 i_exact: str | None = strawberry_django.filter_field()
130 contains: str | None = strawberry_django.filter_field()
131 i_contains: str | None = strawberry_django.filter_field()
132 starts_with: str | None = strawberry_django.filter_field()
133 i_starts_with: str | None = strawberry_django.filter_field()
134 ends_with: str | None = strawberry_django.filter_field()
135 i_ends_with: str | None = strawberry_django.filter_field()
136 in_: list[str] | None = strawberry_django.filter_field()
137 isnull: bool | None = strawberry_django.filter_field()
138 regex: str | None = strawberry_django.filter_field()
139 i_regex: str | None = strawberry_django.filter_field()
142@strawberry.input(one_of=True, description='Lookup for JSON field. Only one of the lookup fields can be set.')
143class JSONLookup:
144 string_lookup: JSONStringLookup | None = strawberry_django.filter_field()
145 int_range_lookup: RangeLookup[int] | None = strawberry_django.filter_field()
146 int_comparison_lookup: ComparisonFilterLookup[int] | None = strawberry_django.filter_field()
147 float_range_lookup: RangeLookup[float] | None = strawberry_django.filter_field()
148 float_comparison_lookup: ComparisonFilterLookup[float] | None = strawberry_django.filter_field()
149 date_lookup: JSONDateFilterLookup | None = strawberry_django.filter_field()
150 datetime_lookup: JSONDatetimeFilterLookup | None = strawberry_django.filter_field()
151 time_lookup: JSONTimeFilterLookup | None = strawberry_django.filter_field()
152 boolean_lookup: FilterLookup[bool] | None = strawberry_django.filter_field()
154 def get_filter(self):
155 for field in self.__strawberry_definition__.fields:
156 value = getattr(self, field.name, None)
157 if value is not strawberry.UNSET:
158 return value
159 return None
162class _NumericLookupMixin:
163 """Shared filter logic for numeric lookup input types (Integer, BigInteger, Float)."""
165 def get_filter(self):
166 for field in self.__strawberry_definition__.fields:
167 value = getattr(self, field.name, None)
168 if value is not strawberry.UNSET:
169 return value
170 return None
172 @strawberry_django.filter_field
173 def filter(self, info: Info, queryset: QuerySet, prefix: DirectiveValue[str] = '') -> tuple[QuerySet, Q]:
174 filters = self.get_filter()
176 if not filters:
177 return queryset, Q()
179 if isinstance(filters, RangeLookup):
180 prefix = f'{prefix}range__'
182 return process_filters(filters=filters, queryset=queryset, info=info, prefix=prefix)
185@strawberry.input(one_of=True, description='Lookup for Integer fields. Only one of the lookup fields can be set.')
186class IntegerLookup(_NumericLookupMixin):
187 filter_lookup: FilterLookup[int] | None = strawberry_django.filter_field()
188 range_lookup: RangeLookup[int] | None = strawberry_django.filter_field()
189 comparison_lookup: ComparisonFilterLookup[int] | None = strawberry_django.filter_field()
192@strawberry.input(one_of=True, description='Lookup for BigInteger fields. Only one of the lookup fields can be set.')
193class BigIntegerLookup(_NumericLookupMixin):
194 filter_lookup: FilterLookup[BigInt] | None = strawberry_django.filter_field()
195 range_lookup: RangeLookup[BigInt] | None = strawberry_django.filter_field()
196 comparison_lookup: ComparisonFilterLookup[BigInt] | None = strawberry_django.filter_field()
199@strawberry.input(one_of=True, description='Lookup for Float fields. Only one of the lookup fields can be set.')
200class FloatLookup(_NumericLookupMixin):
201 filter_lookup: FilterLookup[float] | None = strawberry_django.filter_field()
202 range_lookup: RangeLookup[float] | None = strawberry_django.filter_field()
203 comparison_lookup: ComparisonFilterLookup[float] | None = strawberry_django.filter_field()
206@strawberry.input
207class JSONFilter:
208 """
209 Class for JSON field lookups with paths
210 """
212 path: str
213 lookup: JSONLookup
215 @strawberry_django.filter_field
216 def filter(self, info: Info, queryset: QuerySet, prefix: DirectiveValue[str] = '') -> tuple[QuerySet, Q]:
217 filters = self.lookup.get_filter()
219 if not filters:
220 return queryset, Q()
222 try:
223 safe_path = _validate_json_path(self.path)
224 except ValueError:
225 return queryset, Q()
227 json_path = f'{prefix}{safe_path}__'
228 return process_filters(filters=filters, queryset=queryset, info=info, prefix=json_path)
231@strawberry.enum
232class TreeNodeMatch(Enum):
233 EXACT = 'exact' # Just the node itself
234 DESCENDANTS = 'descendants' # All descendants, excluding the node itself
235 SELF_AND_DESCENDANTS = 'self_and_descendants' # Node and all descendants
236 CHILDREN = 'children' # Just immediate children
237 SIBLINGS = 'siblings' # Nodes with same parent
238 ANCESTORS = 'ancestors' # All parent nodes
239 PARENT = 'parent' # Just immediate parent
242@strawberry.input
243class TreeNodeFilter:
244 id: ID
245 match_type: TreeNodeMatch
247 @strawberry_django.filter_field
248 def filter(self, info: Info, queryset: QuerySet, prefix: DirectiveValue[str] = '') -> tuple[QuerySet, Q]:
249 model_field_name = prefix.removesuffix('__').removesuffix('_id')
250 model_field = None
251 try:
252 model_field = queryset.model._meta.get_field(model_field_name)
253 except FieldDoesNotExist:
254 try:
255 model_field = queryset.model._meta.get_field(f'{model_field_name}s')
256 except FieldDoesNotExist:
257 return queryset, Q(pk__in=[])
259 if hasattr(model_field, 'related_model'):
260 related_model = model_field.related_model
261 else:
262 return queryset, Q(pk__in=[])
264 # Generate base Q filter for the related model without prefix
265 q_filter = generate_tree_node_q_filter(related_model, self)
267 # Handle different relationship types. All variants resolve the related
268 # rows against the q_filter (which may be a compound Q for DESCENDANTS,
269 # ANCESTORS, SIBLINGS, SELF_AND_DESCENDANTS) and join via __in. Destructuring
270 # q_filter.children into kwargs would crash on compound match types.
271 if isinstance(model_field, (ManyToManyField, ManyToManyRel, ForeignKey, ManyToOneRel)):
272 return queryset, Q(**{f'{model_field_name}__in': related_model.objects.filter(q_filter)})
273 return queryset, Q(**{f'{model_field_name}__{k}': v for k, v in q_filter.children})
276def generate_tree_node_q_filter(model_class, filter_value: TreeNodeFilter) -> Q:
277 """
278 Generate Q filter for ltree-backed hierarchical models based on match type.
279 """
280 try:
281 node = model_class.objects.get(id=filter_value.id)
282 except model_class.DoesNotExist:
283 return Q(pk__in=[])
285 if not getattr(node, 'path', None):
286 return Q(id=filter_value.id)
288 if filter_value.match_type == TreeNodeMatch.EXACT:
289 return Q(id=filter_value.id)
290 if filter_value.match_type == TreeNodeMatch.DESCENDANTS:
291 return Q(path__descendant=node.path) & ~Q(id=node.id)
292 if filter_value.match_type == TreeNodeMatch.SELF_AND_DESCENDANTS:
293 return Q(path__descendant_or_equal=node.path)
294 if filter_value.match_type == TreeNodeMatch.CHILDREN:
295 return Q(parent_id=node.id)
296 if filter_value.match_type == TreeNodeMatch.SIBLINGS:
297 return Q(parent_id=node.parent_id) & ~Q(id=node.id)
298 if filter_value.match_type == TreeNodeMatch.ANCESTORS:
299 return Q(path__ancestor=node.path) & ~Q(id=node.id)
300 if filter_value.match_type == TreeNodeMatch.PARENT:
301 return Q(id=node.parent_id) if node.parent_id else Q(pk__in=[])
302 return Q()
305@strawberry.input(one_of=True, description='Lookup for Array fields. Only one of the lookup fields can be set.')
306class ArrayLookup(Generic[T]):
307 """
308 Class for Array field lookups
309 """
311 contains: list[T] | None = strawberry.field(default=strawberry.UNSET, description='Contains the value')
312 contained_by: list[T] | None = strawberry.field(default=strawberry.UNSET, description='Contained by the value')
313 overlap: list[T] | None = strawberry.field(default=strawberry.UNSET, description='Overlaps with the value')
314 length: int | None = strawberry.field(default=strawberry.UNSET, description='Length of the array')
316 @strawberry_django.filter_field
317 def filter(self, info: Info, queryset: QuerySet, prefix: str = '') -> tuple[QuerySet, Q]:
318 # Map the public GraphQL ``length`` field to Django's ``len`` array transform; the
319 # remaining lookups share their name with the corresponding ORM transform.
320 if self.contains is not strawberry.UNSET and self.contains is not None:
321 return queryset, Q(**{f'{prefix}contains': self.contains})
322 if self.contained_by is not strawberry.UNSET and self.contained_by is not None:
323 return queryset, Q(**{f'{prefix}contained_by': self.contained_by})
324 if self.overlap is not strawberry.UNSET and self.overlap is not None:
325 return queryset, Q(**{f'{prefix}overlap': self.overlap})
326 if self.length is not strawberry.UNSET and self.length is not None:
327 return queryset, Q(**{f'{prefix}len': self.length})
328 return queryset, Q()
331@strawberry.input(one_of=True, description='Lookup for Array fields. Only one of the lookup fields can be set.')
332class IntegerArrayLookup(ArrayLookup[int]):
333 pass
336@strawberry.input(one_of=True, description='Lookup for Array fields. Only one of the lookup fields can be set.')
337class FloatArrayLookup(ArrayLookup[float]):
338 pass
341@strawberry.input(one_of=True, description='Lookup for Array fields. Only one of the lookup fields can be set.')
342class StringArrayLookup(ArrayLookup[str]):
343 pass
346@strawberry.input(one_of=True, description='Lookups for an ArrayField(RangeField). Only one may be set.')
347class RangeArrayValueLookup(Generic[T]):
348 """
349 class for Array field of Range fields lookups
350 """
352 contains: T | None = strawberry.field(
353 default=strawberry.UNSET, description='Return rows where any stored range contains this value.'
354 )
356 @strawberry_django.filter_field
357 def filter(self, info: Info, queryset: QuerySet, prefix: str = '') -> tuple[QuerySet, Q]:
358 """
359 Map GraphQL: { <field>: { contains: <T> } } To Django ORM: <field>__range_contains=<T>
360 """
361 if self.contains is strawberry.UNSET or self.contains is None:
362 return queryset, Q()
364 # Build '<prefix>range_contains' so it works for nested paths too
365 return queryset, Q(**{f'{prefix}range_contains': self.contains})
368@strawberry.input(one_of=True, description='Lookups for an ArrayField(IntegerRangeField). Only one may be set.')
369class IntegerRangeArrayLookup(RangeArrayValueLookup[int]):
370 pass