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

1import re 

2from enum import Enum 

3from typing import Generic, TypeVar 

4 

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) 

19 

20from netbox.graphql.scalars import BigInt 

21 

22# ------------------------------------------------------------------ 

23# JSON path validation (VM-323) 

24# ------------------------------------------------------------------ 

25 

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_-]*$') 

30 

31 

32def _validate_json_path(path: str) -> str: 

33 """Validate a JSON traversal path for use in ORM lookups. 

34 

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. 

38 

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") 

47 

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}") 

53 

54 return path 

55 

56 

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) 

71 

72T = TypeVar('T') 

73SKIP_MSG = 'Filter will be skipped on `null` value' 

74 

75 

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 

90 

91 

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 

99 

100 

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 

116 

117 

118@strawberry.input(description='String lookups for JSON field values.') 

119class JSONStringLookup: 

120 """ 

121 String-filter type for use inside JSONLookup. 

122 

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() 

140 

141 

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() 

153 

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 

160 

161 

162class _NumericLookupMixin: 

163 """Shared filter logic for numeric lookup input types (Integer, BigInteger, Float).""" 

164 

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 

171 

172 @strawberry_django.filter_field 

173 def filter(self, info: Info, queryset: QuerySet, prefix: DirectiveValue[str] = '') -> tuple[QuerySet, Q]: 

174 filters = self.get_filter() 

175 

176 if not filters: 

177 return queryset, Q() 

178 

179 if isinstance(filters, RangeLookup): 

180 prefix = f'{prefix}range__' 

181 

182 return process_filters(filters=filters, queryset=queryset, info=info, prefix=prefix) 

183 

184 

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() 

190 

191 

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() 

197 

198 

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() 

204 

205 

206@strawberry.input 

207class JSONFilter: 

208 """ 

209 Class for JSON field lookups with paths 

210 """ 

211 

212 path: str 

213 lookup: JSONLookup 

214 

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() 

218 

219 if not filters: 

220 return queryset, Q() 

221 

222 try: 

223 safe_path = _validate_json_path(self.path) 

224 except ValueError: 

225 return queryset, Q() 

226 

227 json_path = f'{prefix}{safe_path}__' 

228 return process_filters(filters=filters, queryset=queryset, info=info, prefix=json_path) 

229 

230 

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 

240 

241 

242@strawberry.input 

243class TreeNodeFilter: 

244 id: ID 

245 match_type: TreeNodeMatch 

246 

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=[]) 

258 

259 if hasattr(model_field, 'related_model'): 

260 related_model = model_field.related_model 

261 else: 

262 return queryset, Q(pk__in=[]) 

263 

264 # Generate base Q filter for the related model without prefix 

265 q_filter = generate_tree_node_q_filter(related_model, self) 

266 

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}) 

274 

275 

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=[]) 

284 

285 if not getattr(node, 'path', None): 

286 return Q(id=filter_value.id) 

287 

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() 

303 

304 

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 """ 

310 

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') 

315 

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() 

329 

330 

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 

334 

335 

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 

339 

340 

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 

344 

345 

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 """ 

351 

352 contains: T | None = strawberry.field( 

353 default=strawberry.UNSET, description='Return rows where any stored range contains this value.' 

354 ) 

355 

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() 

363 

364 # Build '<prefix>range_contains' so it works for nested paths too 

365 return queryset, Q(**{f'{prefix}range_contains': self.contains}) 

366 

367 

368@strawberry.input(one_of=True, description='Lookups for an ArrayField(IntegerRangeField). Only one may be set.') 

369class IntegerRangeArrayLookup(RangeArrayValueLookup[int]): 

370 pass