Coverage for opt/mealie/lib/python3.12/site-packages/mealie/schema/response/query_filter.py: 72%
404 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-11-25 15:32 +0000
1from __future__ import annotations 1l
3import re 1l
4from collections import deque 1l
5from enum import Enum 1l
6from typing import Any, cast 1l
7from uuid import UUID 1l
9import sqlalchemy as sa 1l
10from dateutil import parser as date_parser 1l
11from dateutil.parser import ParserError 1l
12from humps import decamelize 1l
13from sqlalchemy.ext.associationproxy import AssociationProxyInstance 1l
14from sqlalchemy.orm import InstrumentedAttribute, Mapper 1l
15from sqlalchemy.sql import sqltypes 1l
17from mealie.db.models._model_base import SqlAlchemyBase 1l
18from mealie.db.models._model_utils.datetime import NaiveDateTime 1l
19from mealie.db.models._model_utils.guid import GUID 1l
20from mealie.schema._mealie.mealie_model import MealieModel 1l
23class RelationalKeyword(Enum): 1l
24 IS = "IS" 1l
25 IS_NOT = "IS NOT" 1l
26 IN = "IN" 1l
27 NOT_IN = "NOT IN" 1l
28 CONTAINS_ALL = "CONTAINS ALL" 1l
29 LIKE = "LIKE" 1l
30 NOT_LIKE = "NOT LIKE" 1l
32 @classmethod 1l
33 def parse_component(cls, component: str) -> list[str] | None: 1l
34 """
35 Try to parse a component using a relational keyword
37 If no matching keyword is found, returns None
38 """
40 # extract the attribute name from the component
41 parsed_component = component.split(maxsplit=1) 1dhebfigcja
42 if len(parsed_component) < 2: 1dhebfigcja
43 return None 1dhebfigcja
45 # assume the component has already filtered out the value and try to match a keyword
46 # if we try to filter out the value without checking first, keywords with spaces won't parse correctly
47 possible_keyword = parsed_component[1].strip().lower() 1dhebfigcja
48 for rel_kw in sorted([keyword.value for keyword in cls], key=len, reverse=True): 1dhebfigcja
49 if rel_kw.lower() != possible_keyword: 49 ↛ 52line 49 didn't jump to line 52 because the condition on line 49 was always true1dhebfigcja
50 continue 1dhebfigcja
52 parsed_component[1] = rel_kw
53 return parsed_component
55 # there was no match, so the component may still have the value in it
56 try: 1dhebfigcja
57 _possible_keyword, _value = parsed_component[-1].rsplit(maxsplit=1) 1dhebfigcja
58 parsed_component = [parsed_component[0], _possible_keyword, _value] 1ebfca
59 except ValueError: 1dhebfigcja
60 # the component has no value to filter out
61 return None 1dhebfigcja
63 possible_keyword = parsed_component[1].strip().lower() 1ebfca
64 for rel_kw in sorted([keyword.value for keyword in cls], key=len, reverse=True): 1ebfca
65 if rel_kw.lower() != possible_keyword: 65 ↛ 68line 65 didn't jump to line 68 because the condition on line 65 was always true1ebfca
66 continue 1ebfca
68 parsed_component[1] = rel_kw
69 return parsed_component
71 return None 1ebfca
74class RelationalOperator(Enum): 1l
75 EQ = "=" 1l
76 NOTEQ = "<>" 1l
77 GT = ">" 1l
78 LT = "<" 1l
79 GTE = ">=" 1l
80 LTE = "<=" 1l
82 @classmethod 1l
83 def parse_component(cls, component: str) -> list[str] | None: 1l
84 """
85 Try to parse a component using a relational operator
87 If no matching operator is found, returns None
88 """
90 for rel_op in sorted([operator.value for operator in cls], key=len, reverse=True): 1kdhebfigcja
91 if rel_op not in component: 1kdhebfigcja
92 continue 1kdhebfigcja
94 parsed_component = [base_component.strip() for base_component in component.split(rel_op) if base_component] 1kdhebfigcja
95 parsed_component.insert(1, rel_op) 1kdhebfigcja
96 return parsed_component 1kdhebfigcja
98 return None 1dhebfigcja
101class LogicalOperator(Enum): 1l
102 AND = "AND" 1l
103 OR = "OR" 1l
106class QueryFilterJSONPart(MealieModel): 1l
107 left_parenthesis: str | None = None 1l
108 right_parenthesis: str | None = None 1l
109 logical_operator: LogicalOperator | None = None 1l
111 attribute_name: str | None = None 1l
112 relational_operator: RelationalKeyword | RelationalOperator | None = None 1l
113 value: str | list[str] | None = None 1l
116class QueryFilterJSON(MealieModel): 1l
117 parts: list[QueryFilterJSONPart] = [] 1l
120class QueryFilterBuilderComponent: 1l
121 """A single relational statement"""
123 @staticmethod 1l
124 def strip_quotes_from_string(val: str) -> str: 1l
125 if len(val) > 2 and val[0] == '"' and val[-1] == '"': 125 ↛ 126line 125 didn't jump to line 126 because the condition on line 125 was never true1kdhebfigcja
126 return val[1:-1]
127 else:
128 return val 1kdhebfigcja
130 def __init__( 1l
131 self, attribute_name: str, relationship: RelationalKeyword | RelationalOperator, value: str | list[str]
132 ) -> None:
133 self.attribute_name = decamelize(attribute_name) 1kdhebfigcja
134 self.relationship = relationship 1kdhebfigcja
136 # remove encasing quotes
137 if isinstance(value, str): 137 ↛ 140line 137 didn't jump to line 140 because the condition on line 137 was always true1kdhebfigcja
138 value = self.strip_quotes_from_string(value) 1kdhebfigcja
140 elif isinstance(value, list):
141 value = [self.strip_quotes_from_string(v) for v in value]
143 # validate relationship/value pairs
144 if relationship in [ 144 ↛ 149line 144 didn't jump to line 149 because the condition on line 144 was never true1kdhebfigcja
145 RelationalKeyword.IN,
146 RelationalKeyword.NOT_IN,
147 RelationalKeyword.CONTAINS_ALL,
148 ] and not isinstance(value, list):
149 raise ValueError(
150 f"invalid query string: {relationship.value} must be given a list of values"
151 f"enclosed by {QueryFilterBuilder.l_list_sep} and {QueryFilterBuilder.r_list_sep}"
152 )
154 if relationship is RelationalKeyword.IS or relationship is RelationalKeyword.IS_NOT: 154 ↛ 155line 154 didn't jump to line 155 because the condition on line 154 was never true1kdhebfigcja
155 if not isinstance(value, str) or value.lower() not in ["null", "none"]:
156 raise ValueError(
157 f'invalid query string: "{relationship.value}" can only be used with "NULL", not "{value}"'
158 )
160 self.value = None
161 else:
162 self.value = value 1kdhebfigcja
164 def __repr__(self) -> str: 1l
165 return f"[{self.attribute_name} {self.relationship.value} {self.value}]"
167 def validate(self, model_attr_type: Any) -> Any: 1l
168 """Validate value against an model attribute's type and return a validated value, or raise a ValueError"""
170 sanitized_values: list[Any]
171 if not isinstance(self.value, list): 171 ↛ 174line 171 didn't jump to line 174 because the condition on line 171 was always true1kdhebfigcja
172 sanitized_values = [self.value] 1kdhebfigcja
173 else:
174 sanitized_values = list(self.value)
176 for i, v in enumerate(sanitized_values): 1kdhebfigcja
177 # always allow querying for null values
178 if v is None: 178 ↛ 179line 178 didn't jump to line 179 because the condition on line 178 was never true1kdhebfigcja
179 continue
181 if isinstance(model_attr_type, sqltypes.String): 181 ↛ 182line 181 didn't jump to line 182 because the condition on line 181 was never true1kdhebfigcja
182 sanitized_values[i] = v.lower()
184 if self.relationship is RelationalKeyword.LIKE or self.relationship is RelationalKeyword.NOT_LIKE: 184 ↛ 185line 184 didn't jump to line 185 because the condition on line 184 was never true1kdhebfigcja
185 if not isinstance(model_attr_type, sqltypes.String):
186 raise ValueError(
187 f'invalid query string: "{self.relationship.value}" can only be used with string columns'
188 )
190 if isinstance(model_attr_type, (GUID)): 1kdhebfigcja
191 try: 1kdhebfgcja
192 # we don't set value since a UUID is functionally identical to a string here
193 UUID(v) 1kdhebfgcja
194 except ValueError as e:
195 raise ValueError(f"invalid query string: invalid UUID '{v}'") from e
197 if isinstance(model_attr_type, sqltypes.Date | sqltypes.DateTime | NaiveDateTime): 1kdhebfigcja
198 try: 1kdbigca
199 dt = date_parser.parse(v) 1kdbigca
200 sanitized_values[i] = dt.date() if isinstance(model_attr_type, sqltypes.Date) else dt 1kdbigca
201 except ParserError as e:
202 raise ValueError(f"invalid query string: unknown date or datetime format '{v}'") from e
204 if isinstance(model_attr_type, sqltypes.Boolean): 204 ↛ 205line 204 didn't jump to line 205 because the condition on line 204 was never true1kdhebfigcja
205 try:
206 sanitized_values[i] = v.lower()[0] in ["t", "y"] or v == "1"
207 except IndexError as e:
208 raise ValueError("invalid query string") from e
210 return sanitized_values if isinstance(self.value, list) else sanitized_values[0] 1kdhebfigcja
212 def as_json_model(self) -> QueryFilterJSONPart: 1l
213 return QueryFilterJSONPart(
214 left_parenthesis=None,
215 right_parenthesis=None,
216 logical_operator=None,
217 attribute_name=self.attribute_name,
218 relational_operator=self.relationship,
219 value=self.value,
220 )
223class QueryFilterBuilder: 1l
224 l_group_sep: str = "(" 1l
225 r_group_sep: str = ")" 1l
226 group_seps: set[str] = {l_group_sep, r_group_sep} 1l
228 l_list_sep: str = "[" 1l
229 r_list_sep: str = "]" 1l
230 list_item_sep: str = "," 1l
232 def __init__(self, filter_string: str) -> None: 1l
233 # parse filter string
234 components = QueryFilterBuilder._break_filter_string_into_components(filter_string) 1kdhebfigcja
235 base_components = QueryFilterBuilder._break_components_into_base_components(components) 1kdhebfigcja
236 if base_components.count(QueryFilterBuilder.l_group_sep) != base_components.count( 1kdhebfigcja
237 QueryFilterBuilder.r_group_sep
238 ):
239 raise ValueError("invalid query string: parenthesis are unbalanced") 1debfigcja
241 # parse base components into a filter group
242 self.filter_components = QueryFilterBuilder._parse_base_components_into_filter_components(base_components) 1kdhebfigcja
244 def __repr__(self) -> str: 1l
245 joined = " ".join(
246 [
247 str(component.value if isinstance(component, LogicalOperator) else component)
248 for component in self.filter_components
249 ],
250 )
252 return f"<<{joined}>>"
254 @classmethod 1l
255 def _consolidate_group( 1l
256 cls, group: list[sa.ColumnElement], logical_operators: deque[LogicalOperator]
257 ) -> sa.ColumnElement:
258 consolidated_group_builder: sa.ColumnElement | None = None 1kdhebfigcja
259 for i, element in enumerate(reversed(group)): 1kdhebfigcja
260 if not i: 1kdhebfigcja
261 consolidated_group_builder = element 1kdhebfigcja
262 else:
263 operator = logical_operators.pop() 1da
264 if operator is LogicalOperator.AND: 264 ↛ 266line 264 didn't jump to line 266 because the condition on line 264 was always true1da
265 consolidated_group_builder = sa.and_(consolidated_group_builder, element) 1da
266 elif operator is LogicalOperator.OR:
267 consolidated_group_builder = sa.or_(consolidated_group_builder, element)
268 else:
269 raise ValueError(f"invalid logical operator {operator}")
271 if i == len(group) - 1: 1kdhebfigcja
272 return consolidated_group_builder.self_group() 1kdhebfigcja
274 @classmethod 1l
275 def get_model_and_model_attr_from_attr_string[Model: SqlAlchemyBase]( 1l
276 cls, attr_string: str, model: type[Model], *, query: sa.Select | None = None
277 ) -> tuple[SqlAlchemyBase, InstrumentedAttribute, sa.Select | None]:
278 """
279 Take an attribute string and traverse a database model and its relationships to get the desired
280 model and model attribute. Optionally provide a query to apply the necessary table joins.
282 If the attribute string is invalid, raises a `ValueError`.
284 For instance, the attribute string "user.name" on `RecipeModel`
285 will return the `User` model's `name` attribute.
287 Works with shallow attributes (e.g. "slug" from `RecipeModel`)
288 and arbitrarily deep ones (e.g. "recipe.group.preferences" on `RecipeTimelineEvent`).
289 """
290 mapper: Mapper
291 model_attr: InstrumentedAttribute | None = None 1lkdhebfigcja
293 attribute_chain = decamelize(attr_string).split(".") 1lkdhebfigcja
294 if not attribute_chain: 294 ↛ 295line 294 didn't jump to line 295 because the condition on line 294 was never true1lkdhebfigcja
295 raise ValueError("invalid query string: attribute name cannot be empty")
297 current_model: SqlAlchemyBase = model # type: ignore 1lkdhebfigcja
298 for i, attribute_link in enumerate(attribute_chain): 298 ↛ 331line 298 didn't jump to line 331 because the loop on line 298 didn't complete1lkdhebfigcja
299 try: 1lkdhebfigcja
300 model_attr = getattr(current_model, attribute_link) 1lkdhebfigcja
302 # proxied attributes can't be joined to the query directly, so we need to inspect the proxy
303 # and get the actual model and its attribute
304 if isinstance(model_attr, AssociationProxyInstance): 304 ↛ 305line 304 didn't jump to line 305 because the condition on line 304 was never true1lkdhebfigcja
305 proxied_attribute_link = model_attr.target_collection
306 next_attribute_link = model_attr.value_attr
307 model_attr = getattr(current_model, proxied_attribute_link)
309 if query is not None:
310 query = query.join(model_attr, isouter=True)
312 mapper = sa.inspect(current_model)
313 relationship = mapper.relationships[proxied_attribute_link]
314 current_model = relationship.mapper.class_
315 model_attr = getattr(current_model, next_attribute_link)
317 # at the end of the chain there are no more relationships to inspect
318 if i == len(attribute_chain) - 1: 318 ↛ 321line 318 didn't jump to line 321 because the condition on line 318 was always true1lkdhebfigcja
319 break 1lkdhebfigcja
321 if query is not None:
322 query = query.join(model_attr, isouter=True)
324 mapper = sa.inspect(current_model)
325 relationship = mapper.relationships[attribute_link]
326 current_model = relationship.mapper.class_
328 except (AttributeError, KeyError) as e: 1kdhebfigcja
329 raise ValueError(f"invalid attribute string: '{attr_string}' does not exist on this schema") from e 1kdhebfigcja
331 if model_attr is None: 331 ↛ 332line 331 didn't jump to line 332 because the condition on line 331 was never true1lkdhebfigcja
332 raise ValueError(f"invalid attribute string: '{attr_string}'")
334 return current_model, model_attr, query 1lkdhebfigcja
336 @classmethod 1l
337 def _transform_model_attr(cls, model_attr: InstrumentedAttribute, model_attr_type: Any) -> InstrumentedAttribute: 1l
338 if isinstance(model_attr_type, sqltypes.String): 338 ↛ 339line 338 didn't jump to line 339 because the condition on line 338 was never true1kdhebfigcja
339 model_attr = sa.func.lower(model_attr)
341 return model_attr 1kdhebfigcja
343 @classmethod 1l
344 def _get_filter_element[Model: SqlAlchemyBase]( 1l
345 cls,
346 query: sa.Select,
347 component: QueryFilterBuilderComponent,
348 model: type[Model],
349 model_attr: InstrumentedAttribute,
350 model_attr_type: Any,
351 ) -> sa.ColumnElement:
352 original_model_attr = model_attr 1kdhebfigcja
353 model_attr = cls._transform_model_attr(model_attr, model_attr_type) 1kdhebfigcja
354 value = component.validate(model_attr_type) 1kdhebfigcja
356 # Keywords
357 if component.relationship is RelationalKeyword.IS: 357 ↛ 358line 357 didn't jump to line 358 because the condition on line 357 was never true1kdhebfigcja
358 element = model_attr.is_(value)
359 elif component.relationship is RelationalKeyword.IS_NOT: 359 ↛ 360line 359 didn't jump to line 360 because the condition on line 359 was never true1kdhebfigcja
360 element = model_attr.is_not(value)
361 elif component.relationship is RelationalKeyword.IN: 361 ↛ 362line 361 didn't jump to line 362 because the condition on line 361 was never true1kdhebfigcja
362 element = model_attr.in_(value)
363 elif component.relationship is RelationalKeyword.NOT_IN: 363 ↛ 364line 363 didn't jump to line 364 because the condition on line 363 was never true1kdhebfigcja
364 if original_model_attr.parent.entity != model:
365 subq = query.with_only_columns(model.id).where(model_attr.in_(value))
366 element = sa.not_(model.id.in_(subq))
367 else:
368 element = sa.not_(model_attr.in_(value))
370 elif component.relationship is RelationalKeyword.CONTAINS_ALL: 370 ↛ 371line 370 didn't jump to line 371 because the condition on line 370 was never true1kdhebfigcja
371 if len(value) == 1:
372 element = model_attr.in_(value)
373 else:
374 primary_model_attr: InstrumentedAttribute = getattr(model, component.attribute_name.split(".")[0])
375 element = sa.and_(*(primary_model_attr.any(model_attr == v) for v in value))
376 elif component.relationship is RelationalKeyword.LIKE: 376 ↛ 377line 376 didn't jump to line 377 because the condition on line 376 was never true1kdhebfigcja
377 element = model_attr.ilike(value)
378 elif component.relationship is RelationalKeyword.NOT_LIKE: 378 ↛ 379line 378 didn't jump to line 379 because the condition on line 378 was never true1kdhebfigcja
379 element = model_attr.not_ilike(value)
381 # Operators
382 elif component.relationship is RelationalOperator.EQ: 1kdhebfigcja
383 element = model_attr == value 1kdhebfgcja
384 elif component.relationship is RelationalOperator.NOTEQ: 384 ↛ 385line 384 didn't jump to line 385 because the condition on line 384 was never true1kdbigca
385 element = model_attr != value
386 elif component.relationship is RelationalOperator.GT: 386 ↛ 387line 386 didn't jump to line 387 because the condition on line 386 was never true1kdbigca
387 element = model_attr > value
388 elif component.relationship is RelationalOperator.LT: 388 ↛ 389line 388 didn't jump to line 389 because the condition on line 388 was never true1kdbigca
389 element = model_attr < value
390 elif component.relationship is RelationalOperator.GTE: 1kdbigca
391 element = model_attr >= value 1dbiga
392 elif component.relationship is RelationalOperator.LTE: 392 ↛ 395line 392 didn't jump to line 395 because the condition on line 392 was always true1kdbca
393 element = model_attr <= value 1kdbca
394 else:
395 raise ValueError(f"invalid relationship {component.relationship}")
397 return element 1kdhebfigcja
399 def filter_query[Model: SqlAlchemyBase]( 1l
400 self, query: sa.Select, model: type[Model], column_aliases: dict[str, sa.ColumnElement] | None = None
401 ) -> sa.Select:
402 """
403 Filters a query based on the parsed filter string.
404 If you need to filter on a custom column expression (e.g. a computed property), you can supply column aliases
405 """
406 column_aliases = column_aliases or {} 1kdhebfigcja
408 # join tables and build model chain
409 attr_model_map: dict[int, Any] = {} 1kdhebfigcja
410 model_attr: InstrumentedAttribute
411 for i, component in enumerate(self.filter_components): 1kdhebfigcja
412 if not isinstance(component, QueryFilterBuilderComponent): 1kdhebfigcja
413 continue 1dhebfgca
415 nested_model, model_attr, query = self.get_model_and_model_attr_from_attr_string( 1kdhebfigcja
416 component.attribute_name, model, query=query
417 )
418 attr_model_map[i] = nested_model 1kdhebfigcja
420 # build query filter
421 partial_group: list[sa.ColumnElement] = [] 1kdhebfigcja
422 partial_group_stack: deque[list[sa.ColumnElement]] = deque() 1kdhebfigcja
423 logical_operator_stack: deque[LogicalOperator] = deque() 1kdhebfigcja
424 for i, component in enumerate(self.filter_components): 1kdhebfigcja
425 if component == self.l_group_sep: 1kdhebfigcja
426 partial_group_stack.append(partial_group) 1dhebfgca
427 partial_group = [] 1dhebfgca
429 elif component == self.r_group_sep: 1kdhebfigcja
430 if partial_group: 1dhebfgca
431 complete_group = self._consolidate_group(partial_group, logical_operator_stack) 1ba
432 partial_group = partial_group_stack.pop() 1ba
433 partial_group.append(complete_group) 1ba
434 else:
435 partial_group = partial_group_stack.pop() 1dhebfgca
437 elif isinstance(component, LogicalOperator): 1kdhebfigcja
438 logical_operator_stack.append(component) 1dhebfgca
440 else:
441 component = cast(QueryFilterBuilderComponent, component) 1kdhebfigcja
442 base_attribute_name = component.attribute_name.split(".")[-1] 1kdhebfigcja
443 model_attr = getattr(attr_model_map[i], base_attribute_name) 1kdhebfigcja
445 if (column_alias := column_aliases.get(base_attribute_name)) is not None: 445 ↛ 446line 445 didn't jump to line 446 because the condition on line 445 was never true1kdhebfigcja
446 model_attr = column_alias
448 element = self._get_filter_element(query, component, model, model_attr, model_attr.type) 1kdhebfigcja
449 partial_group.append(element) 1kdhebfigcja
451 # combine the completed groups into one filter
452 while True: 1kdhebfigcja
453 consolidated_group = self._consolidate_group(partial_group, logical_operator_stack) 1kdhebfigcja
454 if not partial_group_stack: 454 ↛ 457line 454 didn't jump to line 457 because the condition on line 454 was always true1kdhebfigcja
455 return query.filter(consolidated_group) 1kdhebfigcja
456 else:
457 partial_group = partial_group_stack.pop()
458 partial_group.append(consolidated_group)
460 @staticmethod 1l
461 def _break_filter_string_into_components(filter_string: str) -> list[str]: 1l
462 """Recursively break filter string into components based on parenthesis groupings"""
463 components = [filter_string] 1kdhebfigcja
464 in_quotes = False 1kdhebfigcja
465 while True: 1kdhebfigcja
466 subcomponents = [] 1kdhebfigcja
467 for component in components: 1kdhebfigcja
468 # don't parse components comprised of only a separator
469 if component in QueryFilterBuilder.group_seps: 1kdhebfigcja
470 subcomponents.append(component) 1dhebfigcja
471 continue 1dhebfigcja
473 # construct a component until it hits the right separator
474 new_component = "" 1kdhebfigcja
475 for c in component: 1kdhebfigcja
476 # ignore characters in-between quotes
477 if c == '"': 1kdhebfigcja
478 in_quotes = not in_quotes 1dhebfica
480 if c in QueryFilterBuilder.group_seps and not in_quotes: 1kdhebfigcja
481 if new_component: 1dhebfigcja
482 subcomponents.append(new_component) 1dhebfigcja
484 subcomponents.append(c) 1dhebfigcja
485 new_component = "" 1dhebfigcja
486 continue 1dhebfigcja
488 new_component += c 1kdhebfigcja
490 if new_component: 1kdhebfigcja
491 subcomponents.append(new_component.strip()) 1kdhebfigcja
493 if components == subcomponents: 1kdhebfigcja
494 break 1kdhebfigcja
496 components = subcomponents 1dhebfigcja
498 return components 1kdhebfigcja
500 @staticmethod 1l
501 def _break_components_into_base_components(components: list[str]) -> list[str | list[str]]: 1l
502 """Further break down components by splitting at relational and logical operators"""
503 pattern = "|".join([f"\\b{operator.value}\\b" for operator in LogicalOperator]) 1kdhebfigcja
504 logical_operators = re.compile(f"({pattern})", flags=re.IGNORECASE) 1kdhebfigcja
506 in_list = False 1kdhebfigcja
507 base_components: list[str | list] = [] 1kdhebfigcja
508 list_value_components = [] 1kdhebfigcja
509 for component in components: 1kdhebfigcja
510 # parse out lists as their own singular sub component
511 subcomponents = component.split(QueryFilterBuilder.l_list_sep) 1kdhebfigcja
512 for i, subcomponent in enumerate(subcomponents): 1kdhebfigcja
513 if not i: 1kdhebfigcja
514 continue 1kdhebfigcja
516 for j, list_value_string in enumerate(subcomponent.split(QueryFilterBuilder.r_list_sep)): 1dhebfigcja
517 if j % 2: 517 ↛ 518line 517 didn't jump to line 518 because the condition on line 517 was never true1dhebfigcja
518 continue
520 list_value_components.append( 1dhebfigcja
521 [val.strip() for val in list_value_string.split(QueryFilterBuilder.list_item_sep)]
522 )
524 quote_offset = 0 1kdhebfigcja
525 subcomponents = component.split('"') 1kdhebfigcja
526 for i, subcomponent in enumerate(subcomponents): 1kdhebfigcja
527 # we are in a list subcomponent, which is already handled
528 if in_list: 1kdhebfigcja
529 if QueryFilterBuilder.r_list_sep in subcomponent: 529 ↛ 531line 529 didn't jump to line 531 because the condition on line 529 was never true1dhebfigcja
530 # filter out the remainder of the list subcomponent and continue parsing
531 base_components.append(list_value_components.pop(0))
532 subcomponent = subcomponent.split(QueryFilterBuilder.r_list_sep, maxsplit=1)[-1].strip()
533 in_list = False
534 else:
535 continue 1dhebfigcja
537 # don't parse components comprised of only a separator
538 if subcomponent in QueryFilterBuilder.group_seps: 1kdhebfigcja
539 quote_offset += 1 1dhebfigcja
540 base_components.append(subcomponent) 1dhebfigcja
541 continue 1dhebfigcja
543 # this subcomponent was surrounded in quotes, so we keep it as-is
544 if (i + quote_offset) % 2: 1kdhebfigcja
545 base_components.append(f'"{subcomponent.strip()}"') 1dhebfica
546 continue 1dhebfica
548 # if the final subcomponent has quotes, it creates an extra empty subcomponent at the end
549 if not subcomponent: 549 ↛ 550line 549 didn't jump to line 550 because the condition on line 549 was never true1kdhebfigcja
550 continue
552 # continue parsing this subcomponent up to the list, then skip over subsequent subcomponents
553 if not in_list and QueryFilterBuilder.l_list_sep in subcomponent: 1kdhebfigcja
554 subcomponent, _new_sub_component = subcomponent.split(QueryFilterBuilder.l_list_sep, maxsplit=1) 1dhebfigcja
555 subcomponent = subcomponent.strip() 1dhebfigcja
556 subcomponents.insert(i + 1, _new_sub_component) 1dhebfigcja
557 quote_offset += 1 1dhebfigcja
558 in_list = True 1dhebfigcja
560 # parse out logical operators
561 new_components = [ 1kdhebfigcja
562 base_component.strip() for base_component in logical_operators.split(subcomponent) if base_component
563 ]
565 # parse out relational keywords and operators
566 # each base_subcomponent has exactly zero or one keyword or operator
567 for component in new_components: 1kdhebfigcja
568 if not component: 568 ↛ 569line 568 didn't jump to line 569 because the condition on line 568 was never true1kdhebfigcja
569 continue
571 # we try relational operators first since they aren't required to be surrounded by spaces
572 parsed_component = RelationalOperator.parse_component(component) 1kdhebfigcja
573 if parsed_component is not None: 1kdhebfigcja
574 base_components.extend(parsed_component) 1kdhebfigcja
575 continue 1kdhebfigcja
577 parsed_component = RelationalKeyword.parse_component(component) 1dhebfigcja
578 if parsed_component is not None: 578 ↛ 579line 578 didn't jump to line 579 because the condition on line 578 was never true1dhebfigcja
579 base_components.extend(parsed_component)
580 continue
582 # this component does not have any keywords or operators, so we just add it as-is
583 base_components.append(component) 1dhebfigcja
585 return base_components 1kdhebfigcja
587 @staticmethod 1l
588 def _parse_base_components_into_filter_components( 1l
589 base_components: list[str | list[str]],
590 ) -> list[str | QueryFilterBuilderComponent | LogicalOperator]:
591 """Walk through base components and construct filter collections"""
592 relational_keywords = [kw.value for kw in RelationalKeyword] 1kdhebfigcja
593 relational_operators = [op.value for op in RelationalOperator] 1kdhebfigcja
594 logical_operators = [op.value for op in LogicalOperator] 1kdhebfigcja
596 # parse QueryFilterComponents and logical operators
597 components: list[str | QueryFilterBuilderComponent | LogicalOperator] = [] 1kdhebfigcja
598 for i, base_component in enumerate(base_components): 1kdhebfigcja
599 if isinstance(base_component, list): 599 ↛ 600line 599 didn't jump to line 600 because the condition on line 599 was never true1kdhebfigcja
600 continue
602 if base_component in QueryFilterBuilder.group_seps: 1kdhebfigcja
603 components.append(base_component) 1dhebfgca
605 elif base_component in relational_keywords or base_component in relational_operators: 1kdhebfigcja
606 relationship: RelationalKeyword | RelationalOperator
607 if base_component in relational_keywords: 607 ↛ 608line 607 didn't jump to line 608 because the condition on line 607 was never true1kdhebfigcja
608 relationship = RelationalKeyword(base_components[i])
609 else:
610 relationship = RelationalOperator(base_components[i]) 1kdhebfigcja
612 components.append( 1kdhebfigcja
613 QueryFilterBuilderComponent(
614 attribute_name=base_components[i - 1], # type: ignore
615 relationship=relationship,
616 value=base_components[i + 1],
617 )
618 )
620 elif base_component.upper() in logical_operators: 1kdhebfigcja
621 components.append(LogicalOperator(base_component.upper())) 1dhebfgca
623 return components 1kdhebfigcja
625 def as_json_model(self) -> QueryFilterJSON: 1l
626 parts: list[QueryFilterJSONPart] = [] 1dhebfigcja
628 current_part: QueryFilterJSONPart | None = None 1dhebfigcja
629 left_parens: list[str] = [] 1dhebfigcja
630 right_parens: list[str] = [] 1dhebfigcja
631 last_logical_operator: LogicalOperator | None = None 1dhebfigcja
633 def add_part(): 1dhebfigcja
634 nonlocal current_part, left_parens, right_parens, last_logical_operator
635 if not current_part: 635 ↛ 638line 635 didn't jump to line 638 because the condition on line 635 was always true1dhebfigcja
636 return 1dhebfigcja
638 current_part.left_parenthesis = "".join(left_parens) or None
639 current_part.right_parenthesis = "".join(right_parens) or None
640 current_part.logical_operator = last_logical_operator
642 parts.append(current_part)
643 current_part = None
644 left_parens.clear()
645 right_parens.clear()
646 last_logical_operator = None
648 for component in self.filter_components: 648 ↛ 649line 648 didn't jump to line 649 because the loop on line 648 never started1dhebfigcja
649 if isinstance(component, QueryFilterBuilderComponent):
650 if current_part:
651 add_part()
652 current_part = component.as_json_model()
654 elif isinstance(component, LogicalOperator):
655 if current_part:
656 add_part()
657 last_logical_operator = component
659 elif isinstance(component, str):
660 if component == QueryFilterBuilder.l_group_sep:
661 left_parens.append(component)
662 elif component == QueryFilterBuilder.r_group_sep:
663 right_parens.append(component)
665 # add last part, if any
666 add_part() 1dhebfigcja
667 return QueryFilterJSON(parts=parts) 1dhebfigcja