Coverage for opt/mealie/lib/python3.12/site-packages/mealie/schema/response/query_filter.py: 73%
404 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 13:45 +0000
1from __future__ import annotations 1n
3import re 1n
4from collections import deque 1n
5from enum import Enum 1n
6from typing import Any, cast 1n
7from uuid import UUID 1n
9import sqlalchemy as sa 1n
10from dateutil import parser as date_parser 1n
11from dateutil.parser import ParserError 1n
12from humps import decamelize 1n
13from sqlalchemy.ext.associationproxy import AssociationProxyInstance 1n
14from sqlalchemy.orm import InstrumentedAttribute, Mapper 1n
15from sqlalchemy.sql import sqltypes 1n
17from mealie.db.models._model_base import SqlAlchemyBase 1n
18from mealie.db.models._model_utils.datetime import NaiveDateTime 1n
19from mealie.db.models._model_utils.guid import GUID 1n
20from mealie.schema._mealie.mealie_model import MealieModel 1n
23class RelationalKeyword(Enum): 1n
24 IS = "IS" 1n
25 IS_NOT = "IS NOT" 1n
26 IN = "IN" 1n
27 NOT_IN = "NOT IN" 1n
28 CONTAINS_ALL = "CONTAINS ALL" 1n
29 LIKE = "LIKE" 1n
30 NOT_LIKE = "NOT LIKE" 1n
32 @classmethod 1n
33 def parse_component(cls, component: str) -> list[str] | None: 1n
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) 1lbdcjfegihkma
42 if len(parsed_component) < 2: 1lbdcjfegihkma
43 return None 1lbdcjfegihkma
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() 1lbdcjfegihka
48 for rel_kw in sorted([keyword.value for keyword in cls], key=len, reverse=True): 1lbdcjfegihka
49 if rel_kw.lower() != possible_keyword: 49 ↛ 52line 49 didn't jump to line 52 because the condition on line 49 was always true1lbdcjfegihka
50 continue 1lbdcjfegihka
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: 1lbdcjfegihka
57 _possible_keyword, _value = parsed_component[-1].rsplit(maxsplit=1) 1lbdcjfegihka
58 parsed_component = [parsed_component[0], _possible_keyword, _value] 1cjfegia
59 except ValueError: 1lbdcjfegihka
60 # the component has no value to filter out
61 return None 1lbdcjfegihka
63 possible_keyword = parsed_component[1].strip().lower() 1cjfegia
64 for rel_kw in sorted([keyword.value for keyword in cls], key=len, reverse=True): 1cjfegia
65 if rel_kw.lower() != possible_keyword: 65 ↛ 68line 65 didn't jump to line 68 because the condition on line 65 was always true1cjfegia
66 continue 1cjfegia
68 parsed_component[1] = rel_kw
69 return parsed_component
71 return None 1cjfegia
74class RelationalOperator(Enum): 1n
75 EQ = "=" 1n
76 NOTEQ = "<>" 1n
77 GT = ">" 1n
78 LT = "<" 1n
79 GTE = ">=" 1n
80 LTE = "<=" 1n
82 @classmethod 1n
83 def parse_component(cls, component: str) -> list[str] | None: 1n
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): 1lbdcjfegihkma
91 if rel_op not in component: 1lbdcjfegihkma
92 continue 1lbdcjfegihkma
94 parsed_component = [base_component.strip() for base_component in component.split(rel_op) if base_component] 1lbdcjfegihkma
95 parsed_component.insert(1, rel_op) 1lbdcjfegihkma
96 return parsed_component 1lbdcjfegihkma
98 return None 1lbdcjfegihkma
101class LogicalOperator(Enum): 1n
102 AND = "AND" 1n
103 OR = "OR" 1n
106class QueryFilterJSONPart(MealieModel): 1n
107 left_parenthesis: str | None = None 1n
108 right_parenthesis: str | None = None 1n
109 logical_operator: LogicalOperator | None = None 1n
111 attribute_name: str | None = None 1n
112 relational_operator: RelationalKeyword | RelationalOperator | None = None 1n
113 value: str | list[str] | None = None 1n
116class QueryFilterJSON(MealieModel): 1n
117 parts: list[QueryFilterJSONPart] = [] 1n
120class QueryFilterBuilderComponent: 1n
121 """A single relational statement"""
123 @staticmethod 1n
124 def strip_quotes_from_string(val: str) -> str: 1n
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 true1lbdcjfegihkma
126 return val[1:-1]
127 else:
128 return val 1lbdcjfegihkma
130 def __init__( 1n
131 self, attribute_name: str, relationship: RelationalKeyword | RelationalOperator, value: str | list[str]
132 ) -> None:
133 self.attribute_name = decamelize(attribute_name) 1lbdcjfegihkma
134 self.relationship = relationship 1lbdcjfegihkma
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 true1lbdcjfegihkma
138 value = self.strip_quotes_from_string(value) 1lbdcjfegihkma
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 true1lbdcjfegihkma
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 true1lbdcjfegihkma
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 1lbdcjfegihkma
164 def __repr__(self) -> str: 1n
165 return f"[{self.attribute_name} {self.relationship.value} {self.value}]"
167 def validate(self, model_attr_type: Any) -> Any: 1n
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 true1lbdcjfegihkma
172 sanitized_values = [self.value] 1lbdcjfegihkma
173 else:
174 sanitized_values = list(self.value)
176 for i, v in enumerate(sanitized_values): 1lbdcjfegihkma
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 true1lbdcjfegihkma
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 true1lbdcjfegihkma
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 true1lbdcjfegihkma
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)): 1lbdcjfegihkma
191 try: 1lbdcjfegihkma
192 # we don't set value since a UUID is functionally identical to a string here
193 UUID(v) 1lbdcjfegihkma
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): 1lbdcjfegihkma
198 try: 1bdcfegiha
199 dt = date_parser.parse(v) 1bdcfegiha
200 sanitized_values[i] = dt.date() if isinstance(model_attr_type, sqltypes.Date) else dt 1bdcfegiha
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): 1lbdcjfegihkma
205 try: 1bcfha
206 sanitized_values[i] = v.lower()[0] in ["t", "y"] or v == "1" 1bcfha
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] 1lbdcjfegihkma
212 def as_json_model(self) -> QueryFilterJSONPart: 1n
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: 1n
224 l_group_sep: str = "(" 1n
225 r_group_sep: str = ")" 1n
226 group_seps: set[str] = {l_group_sep, r_group_sep} 1n
228 l_list_sep: str = "[" 1n
229 r_list_sep: str = "]" 1n
230 list_item_sep: str = "," 1n
232 def __init__(self, filter_string: str) -> None: 1n
233 # parse filter string
234 components = QueryFilterBuilder._break_filter_string_into_components(filter_string) 1lbdcjfegihkma
235 base_components = QueryFilterBuilder._break_components_into_base_components(components) 1lbdcjfegihkma
236 if base_components.count(QueryFilterBuilder.l_group_sep) != base_components.count( 1lbdcjfegihkma
237 QueryFilterBuilder.r_group_sep
238 ):
239 raise ValueError("invalid query string: parenthesis are unbalanced") 1bcjfea
241 # parse base components into a filter group
242 self.filter_components = QueryFilterBuilder._parse_base_components_into_filter_components(base_components) 1lbdcjfegihkma
244 def __repr__(self) -> str: 1n
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 1n
255 def _consolidate_group( 1n
256 cls, group: list[sa.ColumnElement], logical_operators: deque[LogicalOperator]
257 ) -> sa.ColumnElement:
258 consolidated_group_builder: sa.ColumnElement | None = None 1lbdcjfegihkma
259 for i, element in enumerate(reversed(group)): 1lbdcjfegihkma
260 if not i: 1lbdcjfegihkma
261 consolidated_group_builder = element 1lbdcjfegihkma
262 else:
263 operator = logical_operators.pop() 1bcfegha
264 if operator is LogicalOperator.AND: 264 ↛ 266line 264 didn't jump to line 266 because the condition on line 264 was always true1bcfegha
265 consolidated_group_builder = sa.and_(consolidated_group_builder, element) 1bcfegha
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: 1lbdcjfegihkma
272 return consolidated_group_builder.self_group() 1lbdcjfegihkma
274 @classmethod 1n
275 def get_model_and_model_attr_from_attr_string[Model: SqlAlchemyBase]( 1n
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 1nlbdcjfegihkma
293 attribute_chain = decamelize(attr_string).split(".") 1nlbdcjfegihkma
294 if not attribute_chain: 294 ↛ 295line 294 didn't jump to line 295 because the condition on line 294 was never true1nlbdcjfegihkma
295 raise ValueError("invalid query string: attribute name cannot be empty")
297 current_model: SqlAlchemyBase = model # type: ignore 1nlbdcjfegihkma
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 complete1nlbdcjfegihkma
299 try: 1nlbdcjfegihkma
300 model_attr = getattr(current_model, attribute_link) 1nlbdcjfegihkma
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 true1nlbdcjfegihkma
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 true1nlbdcjfegihkma
319 break 1nlbdcjfegihkma
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: 1lbdcjfegihka
329 raise ValueError(f"invalid attribute string: '{attr_string}' does not exist on this schema") from e 1lbdcjfegihka
331 if model_attr is None: 331 ↛ 332line 331 didn't jump to line 332 because the condition on line 331 was never true1nlbdcjfegihkma
332 raise ValueError(f"invalid attribute string: '{attr_string}'")
334 return current_model, model_attr, query 1nlbdcjfegihkma
336 @classmethod 1n
337 def _transform_model_attr(cls, model_attr: InstrumentedAttribute, model_attr_type: Any) -> InstrumentedAttribute: 1n
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 true1lbdcjfegihkma
339 model_attr = sa.func.lower(model_attr)
341 return model_attr 1lbdcjfegihkma
343 @classmethod 1n
344 def _get_filter_element[Model: SqlAlchemyBase]( 1n
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 1lbdcjfegihkma
353 model_attr = cls._transform_model_attr(model_attr, model_attr_type) 1lbdcjfegihkma
354 value = component.validate(model_attr_type) 1lbdcjfegihkma
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 true1lbdcjfegihkma
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 true1lbdcjfegihkma
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 true1lbdcjfegihkma
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 true1lbdcjfegihkma
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 true1lbdcjfegihkma
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 true1lbdcjfegihkma
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 true1lbdcjfegihkma
379 element = model_attr.not_ilike(value)
381 # Operators
382 elif component.relationship is RelationalOperator.EQ: 1lbdcjfegihkma
383 element = model_attr == value 1lbdcjfegihkma
384 elif component.relationship is RelationalOperator.NOTEQ: 384 ↛ 385line 384 didn't jump to line 385 because the condition on line 384 was never true1bdcfegiha
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 true1bdcfegiha
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 true1bdcfegiha
389 element = model_attr < value
390 elif component.relationship is RelationalOperator.GTE: 1bdcfegiha
391 element = model_attr >= value 1bdcfegiha
392 elif component.relationship is RelationalOperator.LTE: 392 ↛ 395line 392 didn't jump to line 395 because the condition on line 392 was always true1bdcfegha
393 element = model_attr <= value 1bdcfegha
394 else:
395 raise ValueError(f"invalid relationship {component.relationship}")
397 return element 1lbdcjfegihkma
399 def filter_query[Model: SqlAlchemyBase]( 1n
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 {} 1lbdcjfegihkma
408 # join tables and build model chain
409 attr_model_map: dict[int, Any] = {} 1lbdcjfegihkma
410 model_attr: InstrumentedAttribute
411 for i, component in enumerate(self.filter_components): 1lbdcjfegihkma
412 if not isinstance(component, QueryFilterBuilderComponent): 1lbdcjfegihkma
413 continue 1bdcjfegihka
415 nested_model, model_attr, query = self.get_model_and_model_attr_from_attr_string( 1lbdcjfegihkma
416 component.attribute_name, model, query=query
417 )
418 attr_model_map[i] = nested_model 1lbdcjfegihkma
420 # build query filter
421 partial_group: list[sa.ColumnElement] = [] 1lbdcjfegihkma
422 partial_group_stack: deque[list[sa.ColumnElement]] = deque() 1lbdcjfegihkma
423 logical_operator_stack: deque[LogicalOperator] = deque() 1lbdcjfegihkma
424 for i, component in enumerate(self.filter_components): 1lbdcjfegihkma
425 if component == self.l_group_sep: 1lbdcjfegihkma
426 partial_group_stack.append(partial_group) 1bdcjfegihka
427 partial_group = [] 1bdcjfegihka
429 elif component == self.r_group_sep: 1lbdcjfegihkma
430 if partial_group: 1bdcjfegihka
431 complete_group = self._consolidate_group(partial_group, logical_operator_stack) 1bdcfegha
432 partial_group = partial_group_stack.pop() 1bdcfegha
433 partial_group.append(complete_group) 1bdcfegha
434 else:
435 partial_group = partial_group_stack.pop() 1bdcjfegihka
437 elif isinstance(component, LogicalOperator): 1lbdcjfegihkma
438 logical_operator_stack.append(component) 1bdcjfegihka
440 else:
441 component = cast(QueryFilterBuilderComponent, component) 1lbdcjfegihkma
442 base_attribute_name = component.attribute_name.split(".")[-1] 1lbdcjfegihkma
443 model_attr = getattr(attr_model_map[i], base_attribute_name) 1lbdcjfegihkma
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 true1lbdcjfegihkma
446 model_attr = column_alias
448 element = self._get_filter_element(query, component, model, model_attr, model_attr.type) 1lbdcjfegihkma
449 partial_group.append(element) 1lbdcjfegihkma
451 # combine the completed groups into one filter
452 while True: 1lbdcjfegihkma
453 consolidated_group = self._consolidate_group(partial_group, logical_operator_stack) 1lbdcjfegihkma
454 if not partial_group_stack: 454 ↛ 457line 454 didn't jump to line 457 because the condition on line 454 was always true1lbdcjfegihkma
455 return query.filter(consolidated_group) 1lbdcjfegihkma
456 else:
457 partial_group = partial_group_stack.pop()
458 partial_group.append(consolidated_group)
460 @staticmethod 1n
461 def _break_filter_string_into_components(filter_string: str) -> list[str]: 1n
462 """Recursively break filter string into components based on parenthesis groupings"""
463 components = [filter_string] 1lbdcjfegihkma
464 in_quotes = False 1lbdcjfegihkma
465 while True: 1lbdcjfegihkma
466 subcomponents = [] 1lbdcjfegihkma
467 for component in components: 1lbdcjfegihkma
468 # don't parse components comprised of only a separator
469 if component in QueryFilterBuilder.group_seps: 1lbdcjfegihkma
470 subcomponents.append(component) 1bdcjfegihka
471 continue 1bdcjfegihka
473 # construct a component until it hits the right separator
474 new_component = "" 1lbdcjfegihkma
475 for c in component: 1lbdcjfegihkma
476 # ignore characters in-between quotes
477 if c == '"': 1lbdcjfegihkma
478 in_quotes = not in_quotes 1lbdcjegika
480 if c in QueryFilterBuilder.group_seps and not in_quotes: 1lbdcjfegihkma
481 if new_component: 1bdcjfegihka
482 subcomponents.append(new_component) 1bdcjfegihka
484 subcomponents.append(c) 1bdcjfegihka
485 new_component = "" 1bdcjfegihka
486 continue 1bdcjfegihka
488 new_component += c 1lbdcjfegihkma
490 if new_component: 1lbdcjfegihkma
491 subcomponents.append(new_component.strip()) 1lbdcjfegihkma
493 if components == subcomponents: 1lbdcjfegihkma
494 break 1lbdcjfegihkma
496 components = subcomponents 1lbdcjfegihka
498 return components 1lbdcjfegihkma
500 @staticmethod 1n
501 def _break_components_into_base_components(components: list[str]) -> list[str | list[str]]: 1n
502 """Further break down components by splitting at relational and logical operators"""
503 pattern = "|".join([f"\\b{operator.value}\\b" for operator in LogicalOperator]) 1lbdcjfegihkma
504 logical_operators = re.compile(f"({pattern})", flags=re.IGNORECASE) 1lbdcjfegihkma
506 in_list = False 1lbdcjfegihkma
507 base_components: list[str | list] = [] 1lbdcjfegihkma
508 list_value_components = [] 1lbdcjfegihkma
509 for component in components: 1lbdcjfegihkma
510 # parse out lists as their own singular sub component
511 subcomponents = component.split(QueryFilterBuilder.l_list_sep) 1lbdcjfegihkma
512 for i, subcomponent in enumerate(subcomponents): 1lbdcjfegihkma
513 if not i: 1lbdcjfegihkma
514 continue 1lbdcjfegihkma
516 for j, list_value_string in enumerate(subcomponent.split(QueryFilterBuilder.r_list_sep)): 1bda
517 if j % 2: 517 ↛ 518line 517 didn't jump to line 518 because the condition on line 517 was never true1bda
518 continue
520 list_value_components.append( 1bda
521 [val.strip() for val in list_value_string.split(QueryFilterBuilder.list_item_sep)]
522 )
524 quote_offset = 0 1lbdcjfegihkma
525 subcomponents = component.split('"') 1lbdcjfegihkma
526 for i, subcomponent in enumerate(subcomponents): 1lbdcjfegihkma
527 # we are in a list subcomponent, which is already handled
528 if in_list: 1lbdcjfegihkma
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 true1bda
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 1bda
537 # don't parse components comprised of only a separator
538 if subcomponent in QueryFilterBuilder.group_seps: 1lbdcjfegihkma
539 quote_offset += 1 1bdcjfegihka
540 base_components.append(subcomponent) 1bdcjfegihka
541 continue 1bdcjfegihka
543 # this subcomponent was surrounded in quotes, so we keep it as-is
544 if (i + quote_offset) % 2: 1lbdcjfegihkma
545 base_components.append(f'"{subcomponent.strip()}"') 1lbdcjegika
546 continue 1lbdcjegika
548 # if the final subcomponent has quotes, it creates an extra empty subcomponent at the end
549 if not subcomponent: 1lbdcjfegihkma
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: 1lbdcjfegihkma
554 subcomponent, _new_sub_component = subcomponent.split(QueryFilterBuilder.l_list_sep, maxsplit=1) 1bda
555 subcomponent = subcomponent.strip() 1bda
556 subcomponents.insert(i + 1, _new_sub_component) 1bda
557 quote_offset += 1 1bda
558 in_list = True 1bda
560 # parse out logical operators
561 new_components = [ 1lbdcjfegihkma
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: 1lbdcjfegihkma
568 if not component: 568 ↛ 569line 568 didn't jump to line 569 because the condition on line 568 was never true1lbdcjfegihkma
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) 1lbdcjfegihkma
573 if parsed_component is not None: 1lbdcjfegihkma
574 base_components.extend(parsed_component) 1lbdcjfegihkma
575 continue 1lbdcjfegihkma
577 parsed_component = RelationalKeyword.parse_component(component) 1lbdcjfegihkma
578 if parsed_component is not None: 578 ↛ 579line 578 didn't jump to line 579 because the condition on line 578 was never true1lbdcjfegihkma
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) 1lbdcjfegihkma
585 return base_components 1lbdcjfegihkma
587 @staticmethod 1n
588 def _parse_base_components_into_filter_components( 1n
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] 1lbdcjfegihkma
593 relational_operators = [op.value for op in RelationalOperator] 1lbdcjfegihkma
594 logical_operators = [op.value for op in LogicalOperator] 1lbdcjfegihkma
596 # parse QueryFilterComponents and logical operators
597 components: list[str | QueryFilterBuilderComponent | LogicalOperator] = [] 1lbdcjfegihkma
598 for i, base_component in enumerate(base_components): 1lbdcjfegihkma
599 if isinstance(base_component, list): 599 ↛ 600line 599 didn't jump to line 600 because the condition on line 599 was never true1lbdcjfegihkma
600 continue
602 if base_component in QueryFilterBuilder.group_seps: 1lbdcjfegihkma
603 components.append(base_component) 1bdcjfegihka
605 elif base_component in relational_keywords or base_component in relational_operators: 1lbdcjfegihkma
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 true1lbdcjfegihkma
608 relationship = RelationalKeyword(base_components[i])
609 else:
610 relationship = RelationalOperator(base_components[i]) 1lbdcjfegihkma
612 components.append( 1lbdcjfegihkma
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: 1lbdcjfegihkma
621 components.append(LogicalOperator(base_component.upper())) 1bdcjfegihka
623 return components 1lbdcjfegihkma
625 def as_json_model(self) -> QueryFilterJSON: 1n
626 parts: list[QueryFilterJSONPart] = [] 1lbdcjfegihka
628 current_part: QueryFilterJSONPart | None = None 1lbdcjfegihka
629 left_parens: list[str] = [] 1lbdcjfegihka
630 right_parens: list[str] = [] 1lbdcjfegihka
631 last_logical_operator: LogicalOperator | None = None 1lbdcjfegihka
633 def add_part(): 1lbdcjfegihka
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 true1lbdcjfegihka
636 return 1lbdcjfegihka
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 started1lbdcjfegihka
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() 1lbdcjfegihka
667 return QueryFilterJSON(parts=parts) 1lbdcjfegihka