Coverage for opt/mealie/lib/python3.12/site-packages/mealie/schema/response/query_filter.py: 74%

1from __future__ import annotations 1 ctx1j

2 

3import re 1 ctx1j

4from collections import deque 1 ctx1j

5from enum import Enum 1 ctx1j

6from typing import Any, cast 1 ctx1j

7from uuid import UUID 1 ctx1j

8 

9import sqlalchemy as sa 1 ctx1j

10from dateutil import parser as date_parser 1 ctx1j

11from dateutil.parser import ParserError 1 ctx1j

12from humps import decamelize 1 ctx1j

13from sqlalchemy.ext.associationproxy import AssociationProxyInstance 1 ctx1j

14from sqlalchemy.orm import InstrumentedAttribute, Mapper 1 ctx1j

15from sqlalchemy.sql import sqltypes 1 ctx1j

16 

17from mealie.db.models._model_base import SqlAlchemyBase 1 ctx1j

18from mealie.db.models._model_utils.datetime import NaiveDateTime 1 ctx1j

19from mealie.db.models._model_utils.guid import GUID 1 ctx1j

20from mealie.schema._mealie.mealie_model import MealieModel 1 ctx1j

21 

22 

23class RelationalKeyword(Enum): 1 ctx1j

24 IS = "IS" 1 ctx1j

25 IS_NOT = "IS NOT" 1 ctx1j

26 IN = "IN" 1 ctx1j

27 NOT_IN = "NOT IN" 1 ctx1j

28 CONTAINS_ALL = "CONTAINS ALL" 1 ctx1j

29 LIKE = "LIKE" 1 ctx1j

30 NOT_LIKE = "NOT LIKE" 1 ctx1j

31 

32 @classmethod 1 ctx1j

33 def parse_component(cls, component: str) -> list[str] | None: 1 ctx1j

34 """ 

35 Try to parse a component using a relational keyword 

36 

37 If no matching keyword is found, returns None 

38 """ 

39 

40 # extract the attribute name from the component 

41 parsed_component = component.split(maxsplit=1) 9 ctx1gihfebcda

42 if len(parsed_component) < 2: 9 ctx1gihfebcda

43 return None 9 ctx1gihfebcda

44 

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

48 for rel_kw in sorted([keyword.value for keyword in cls], key=len, reverse=True): 6 ctx1gihfea

49 if rel_kw.lower() != possible_keyword: 49 ↛ 52line 49 didn't jump to line 52 because the condition on line 49 was always true6 ctx1gihfea

50 continue 6 ctx1gihfea

51 

52 parsed_component[1] = rel_kw 

53 return parsed_component 

54 

55 # there was no match, so the component may still have the value in it 

56 try: 6 ctx1gihfea

57 _possible_keyword, _value = parsed_component[-1].rsplit(maxsplit=1) 6 ctx1gihfea

58 parsed_component = [parsed_component[0], _possible_keyword, _value] 4 ctx1ifea

59 except ValueError: 6 ctx1gihfea

60 # the component has no value to filter out 

61 return None 6 ctx1gihfea

62 

63 possible_keyword = parsed_component[1].strip().lower() 4 ctx1ifea

64 for rel_kw in sorted([keyword.value for keyword in cls], key=len, reverse=True): 4 ctx1ifea

65 if rel_kw.lower() != possible_keyword: 65 ↛ 68line 65 didn't jump to line 68 because the condition on line 65 was always true4 ctx1ifea

66 continue 4 ctx1ifea

67 

68 parsed_component[1] = rel_kw 

69 return parsed_component 

70 

71 return None 4 ctx1ifea

72 

73 

74class RelationalOperator(Enum): 1 ctx1j

75 EQ = "=" 1 ctx1j

76 NOTEQ = "<>" 1 ctx1j

77 GT = ">" 1 ctx1j

78 LT = "<" 1 ctx1j

79 GTE = ">=" 1 ctx1j

80 LTE = "<=" 1 ctx1j

81 

82 @classmethod 1 ctx1j

83 def parse_component(cls, component: str) -> list[str] | None: 1 ctx1j

84 """ 

85 Try to parse a component using a relational operator 

86 

87 If no matching operator is found, returns None 

88 """ 

89 

90 for rel_op in sorted([operator.value for operator in cls], key=len, reverse=True): 9 ctx1gihfebcda

91 if rel_op not in component: 9 ctx1gihfebcda

92 continue 9 ctx1gihfebcda

93 

94 parsed_component = [base_component.strip() for base_component in component.split(rel_op) if base_component] 8 ctx1ghfebcda

95 parsed_component.insert(1, rel_op) 8 ctx1ghfebcda

96 return parsed_component 8 ctx1ghfebcda

97 

98 return None 9 ctx1gihfebcda

99 

100 

101class LogicalOperator(Enum): 1 ctx1j

102 AND = "AND" 1 ctx1j

103 OR = "OR" 1 ctx1j

104 

105 

106class QueryFilterJSONPart(MealieModel): 1 ctx1j

107 left_parenthesis: str | None = None 1 ctx1j

108 right_parenthesis: str | None = None 1 ctx1j

109 logical_operator: LogicalOperator | None = None 1 ctx1j

110 

111 attribute_name: str | None = None 1 ctx1j

112 relational_operator: RelationalKeyword | RelationalOperator | None = None 1 ctx1j

113 value: str | list[str] | None = None 1 ctx1j

114 

115 

116class QueryFilterJSON(MealieModel): 1 ctx1j

117 parts: list[QueryFilterJSONPart] = [] 1 ctx1j

118 

119 

120class QueryFilterBuilderComponent: 1 ctx1j

121 """A single relational statement""" 

122 

123 @staticmethod 1 ctx1j

124 def strip_quotes_from_string(val: str) -> str: 1 ctx1j

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 true7 ctx1gfebcda

126 return val[1:-1] 

127 else: 

128 return val 7 ctx1gfebcda

129 

130 def __init__( 1 ctx1j

131 self, attribute_name: str, relationship: RelationalKeyword | RelationalOperator, value: str | list[str] 

132 ) -> None: 

133 self.attribute_name = decamelize(attribute_name) 7 ctx1gfebcda

134 self.relationship = relationship 7 ctx1gfebcda

135 

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 true7 ctx1gfebcda

138 value = self.strip_quotes_from_string(value) 7 ctx1gfebcda

139 

140 elif isinstance(value, list): 

141 value = [self.strip_quotes_from_string(v) for v in value] 

142 

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 true7 ctx1gfebcda

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 ) 

153 

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 true7 ctx1gfebcda

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 ) 

159 

160 self.value = None 

161 else: 

162 self.value = value 7 ctx1gfebcda

163 

164 def __repr__(self) -> str: 1 ctx1j

165 return f"[{self.attribute_name} {self.relationship.value} {self.value}]" 

166 

167 def validate(self, model_attr_type: Any) -> Any: 1 ctx1j

168 """Validate value against an model attribute's type and return a validated value, or raise a ValueError""" 

169 

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 true4 ctx1bcda

172 sanitized_values = [self.value] 4 ctx1bcda

173 else: 

174 sanitized_values = list(self.value) 

175 

176 for i, v in enumerate(sanitized_values): 4 ctx1bcda

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 true4 ctx1bcda

179 continue 

180 

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 true4 ctx1bcda

182 sanitized_values[i] = v.lower() 

183 

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 true4 ctx1bcda

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 ) 

189 

190 if isinstance(model_attr_type, (GUID)): 4 ctx1bcda

191 try: 4 ctx1bcda

192 # we don't set value since a UUID is functionally identical to a string here 

193 UUID(v) 4 ctx1bcda

194 except ValueError as e: 

195 raise ValueError(f"invalid query string: invalid UUID '{v}'") from e 

196 

197 if isinstance(model_attr_type, sqltypes.Date | sqltypes.DateTime | NaiveDateTime): 4 ctx1bcda

198 try: (empty)

199 dt = date_parser.parse(v) (empty)

200 sanitized_values[i] = dt.date() if isinstance(model_attr_type, sqltypes.Date) else dt (empty)

201 except ParserError as e: 

202 raise ValueError(f"invalid query string: unknown date or datetime format '{v}'") from e 

203 

204 if isinstance(model_attr_type, sqltypes.Boolean): 4 ctx1bcda

205 try: 4 ctx1bcda

206 sanitized_values[i] = v.lower()[0] in ["t", "y"] or v == "1" 4 ctx1bcda

207 except IndexError as e: 

208 raise ValueError("invalid query string") from e 

209 

210 return sanitized_values if isinstance(self.value, list) else sanitized_values[0] 4 ctx1bcda

211 

212 def as_json_model(self) -> QueryFilterJSONPart: 1 ctx1j

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 ) 

221 

222 

223class QueryFilterBuilder: 1 ctx1j

224 l_group_sep: str = "(" 1 ctx1j

225 r_group_sep: str = ")" 1 ctx1j

226 group_seps: set[str] = {l_group_sep, r_group_sep} 1 ctx1j

227 

228 l_list_sep: str = "[" 1 ctx1j

229 r_list_sep: str = "]" 1 ctx1j

230 list_item_sep: str = "," 1 ctx1j

231 

232 def __init__(self, filter_string: str) -> None: 1 ctx1j

233 # parse filter string 

234 components = QueryFilterBuilder._break_filter_string_into_components(filter_string) 9 ctx1gihfebcda

235 base_components = QueryFilterBuilder._break_components_into_base_components(components) 9 ctx1gihfebcda

236 if base_components.count(QueryFilterBuilder.l_group_sep) != base_components.count( 9 ctx1gihfebcda

237 QueryFilterBuilder.r_group_sep 

238 ): 

239 raise ValueError("invalid query string: parenthesis are unbalanced") 2 ctx1ea

240 

241 # parse base components into a filter group 

242 self.filter_components = QueryFilterBuilder._parse_base_components_into_filter_components(base_components) 9 ctx1gihfebcda

243 

244 def __repr__(self) -> str: 1 ctx1j

245 joined = " ".join( 

246 [ 

247 str(component.value if isinstance(component, LogicalOperator) else component) 

248 for component in self.filter_components 

249 ], 

250 ) 

251 

252 return f"<<{joined}>>" 

253 

254 @classmethod 1 ctx1j

255 def _consolidate_group( 1 ctx1j

256 cls, group: list[sa.ColumnElement], logical_operators: deque[LogicalOperator] 

257 ) -> sa.ColumnElement: 

258 consolidated_group_builder: sa.ColumnElement | None = None 9 ctx1gihfebcda

259 for i, element in enumerate(reversed(group)): 9 ctx1gihfebcda

260 if not i: 4 ctx1bcda

261 consolidated_group_builder = element 4 ctx1bcda

262 else: 

263 operator = logical_operators.pop() 4 ctx1bcda

264 if operator is LogicalOperator.AND: 264 ↛ 266line 264 didn't jump to line 266 because the condition on line 264 was always true4 ctx1bcda

265 consolidated_group_builder = sa.and_(consolidated_group_builder, element) 4 ctx1bcda

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

270 

271 if i == len(group) - 1: 4 ctx1bcda

272 return consolidated_group_builder.self_group() 4 ctx1bcda

273 

274 @classmethod 1 ctx1j

275 def get_model_and_model_attr_from_attr_string[Model: SqlAlchemyBase]( 1 ctx1j

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. 

281 

282 If the attribute string is invalid, raises a `ValueError`. 

283 

284 For instance, the attribute string "user.name" on `RecipeModel` 

285 will return the `User` model's `name` attribute. 

286 

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 16 ctx1jgihfebklcdmnopa

292 

293 attribute_chain = decamelize(attr_string).split(".") 16 ctx1jgihfebklcdmnopa

294 if not attribute_chain: 294 ↛ 295line 294 didn't jump to line 295 because the condition on line 294 was never true16 ctx1jgihfebklcdmnopa

295 raise ValueError("invalid query string: attribute name cannot be empty") 

296 

297 current_model: SqlAlchemyBase = model # type: ignore 16 ctx1jgihfebklcdmnopa

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 complete16 ctx1jgihfebklcdmnopa

299 try: 16 ctx1jgihfebklcdmnopa

300 model_attr = getattr(current_model, attribute_link) 16 ctx1jgihfebklcdmnopa

301 

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 true16 ctx1jgihfebklcdmnopa

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) 

308 

309 if query is not None: 

310 query = query.join(model_attr, isouter=True) 

311 

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) 

316 

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 true16 ctx1jgihfebklcdmnopa

319 break 16 ctx1jgihfebklcdmnopa

320 

321 if query is not None: 

322 query = query.join(model_attr, isouter=True) 

323 

324 mapper = sa.inspect(current_model) 

325 relationship = mapper.relationships[attribute_link] 

326 current_model = relationship.mapper.class_ 

327 

328 except (AttributeError, KeyError) as e: 6 ctx1gihfea

329 raise ValueError(f"invalid attribute string: '{attr_string}' does not exist on this schema") from e 6 ctx1gihfea

330 

331 if model_attr is None: 331 ↛ 332line 331 didn't jump to line 332 because the condition on line 331 was never true16 ctx1jgihfebklcdmnopa

332 raise ValueError(f"invalid attribute string: '{attr_string}'") 

333 

334 return current_model, model_attr, query 16 ctx1jgihfebklcdmnopa

335 

336 @classmethod 1 ctx1j

337 def _transform_model_attr(cls, model_attr: InstrumentedAttribute, model_attr_type: Any) -> InstrumentedAttribute: 1 ctx1j

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 true4 ctx1bcda

339 model_attr = sa.func.lower(model_attr) 

340 

341 return model_attr 4 ctx1bcda

342 

343 @classmethod 1 ctx1j

344 def _get_filter_element[Model: SqlAlchemyBase]( 1 ctx1j

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 4 ctx1bcda

353 model_attr = cls._transform_model_attr(model_attr, model_attr_type) 4 ctx1bcda

354 value = component.validate(model_attr_type) 4 ctx1bcda

355 

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 true4 ctx1bcda

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 true4 ctx1bcda

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 true4 ctx1bcda

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 true4 ctx1bcda

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

369 

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 true4 ctx1bcda

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 true4 ctx1bcda

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 true4 ctx1bcda

379 element = model_attr.not_ilike(value) 

380 

381 # Operators 

382 elif component.relationship is RelationalOperator.EQ: 4 ctx1bcda

383 element = model_attr == value 4 ctx1bcda

384 elif component.relationship is RelationalOperator.NOTEQ: 384 ↛ 385line 384 didn't jump to line 385 because the condition on line 384 was never true(empty)

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

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

389 element = model_attr < value 

390 elif component.relationship is RelationalOperator.GTE: (empty)

391 element = model_attr >= value (empty)

392 elif component.relationship is RelationalOperator.LTE: 392 ↛ 395line 392 didn't jump to line 395 because the condition on line 392 was always true(empty)

393 element = model_attr <= value (empty)

394 else: 

395 raise ValueError(f"invalid relationship {component.relationship}") 

396 

397 return element 4 ctx1bcda

398 

399 def filter_query[Model: SqlAlchemyBase]( 1 ctx1j

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 {} 9 ctx1gihfebcda

407 

408 # join tables and build model chain 

409 attr_model_map: dict[int, Any] = {} 9 ctx1gihfebcda

410 model_attr: InstrumentedAttribute 

411 for i, component in enumerate(self.filter_components): 9 ctx1gihfebcda

412 if not isinstance(component, QueryFilterBuilderComponent): 7 ctx1gfebcda

413 continue 4 ctx1bcda

414 

415 nested_model, model_attr, query = self.get_model_and_model_attr_from_attr_string( 7 ctx1gfebcda

416 component.attribute_name, model, query=query 

417 ) 

418 attr_model_map[i] = nested_model 4 ctx1bcda

419 

420 # build query filter 

421 partial_group: list[sa.ColumnElement] = [] 9 ctx1gihfebcda

422 partial_group_stack: deque[list[sa.ColumnElement]] = deque() 9 ctx1gihfebcda

423 logical_operator_stack: deque[LogicalOperator] = deque() 9 ctx1gihfebcda

424 for i, component in enumerate(self.filter_components): 9 ctx1gihfebcda

425 if component == self.l_group_sep: 4 ctx1bcda

426 partial_group_stack.append(partial_group) (empty)

427 partial_group = [] (empty)

428 

429 elif component == self.r_group_sep: 4 ctx1bcda

430 if partial_group: (empty)

431 complete_group = self._consolidate_group(partial_group, logical_operator_stack) (empty)

432 partial_group = partial_group_stack.pop() (empty)

433 partial_group.append(complete_group) (empty)

434 else: 

435 partial_group = partial_group_stack.pop() (empty)

436 

437 elif isinstance(component, LogicalOperator): 4 ctx1bcda

438 logical_operator_stack.append(component) 4 ctx1bcda

439 

440 else: 

441 component = cast(QueryFilterBuilderComponent, component) 4 ctx1bcda

442 base_attribute_name = component.attribute_name.split(".")[-1] 4 ctx1bcda

443 model_attr = getattr(attr_model_map[i], base_attribute_name) 4 ctx1bcda

444 

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 true4 ctx1bcda

446 model_attr = column_alias 

447 

448 element = self._get_filter_element(query, component, model, model_attr, model_attr.type) 4 ctx1bcda

449 partial_group.append(element) 4 ctx1bcda

450 

451 # combine the completed groups into one filter 

452 while True: 9 ctx1gihfebcda

453 consolidated_group = self._consolidate_group(partial_group, logical_operator_stack) 9 ctx1gihfebcda

454 if not partial_group_stack: 454 ↛ 457line 454 didn't jump to line 457 because the condition on line 454 was always true9 ctx1gihfebcda

455 return query.filter(consolidated_group) 9 ctx1gihfebcda

456 else: 

457 partial_group = partial_group_stack.pop() 

458 partial_group.append(consolidated_group) 

459 

460 @staticmethod 1 ctx1j

461 def _break_filter_string_into_components(filter_string: str) -> list[str]: 1 ctx1j

462 """Recursively break filter string into components based on parenthesis groupings""" 

463 components = [filter_string] 9 ctx1gihfebcda

464 in_quotes = False 9 ctx1gihfebcda

465 while True: 9 ctx1gihfebcda

466 subcomponents = [] 9 ctx1gihfebcda

467 for component in components: 9 ctx1gihfebcda

468 # don't parse components comprised of only a separator 

469 if component in QueryFilterBuilder.group_seps: 9 ctx1gihfebcda

470 subcomponents.append(component) 2 ctx1ea

471 continue 2 ctx1ea

472 

473 # construct a component until it hits the right separator 

474 new_component = "" 9 ctx1gihfebcda

475 for c in component: 9 ctx1gihfebcda

476 # ignore characters in-between quotes 

477 if c == '"': 9 ctx1gihfebcda

478 in_quotes = not in_quotes 2 ctx1ea

479 

480 if c in QueryFilterBuilder.group_seps and not in_quotes: 9 ctx1gihfebcda

481 if new_component: 2 ctx1ea

482 subcomponents.append(new_component) 2 ctx1ea

483 

484 subcomponents.append(c) 2 ctx1ea

485 new_component = "" 2 ctx1ea

486 continue 2 ctx1ea

487 

488 new_component += c 9 ctx1gihfebcda

489 

490 if new_component: 9 ctx1gihfebcda

491 subcomponents.append(new_component.strip()) 9 ctx1gihfebcda

492 

493 if components == subcomponents: 9 ctx1gihfebcda

494 break 9 ctx1gihfebcda

495 

496 components = subcomponents 5 ctx1ghfea

497 

498 return components 9 ctx1gihfebcda

499 

500 @staticmethod 1 ctx1j

501 def _break_components_into_base_components(components: list[str]) -> list[str | list[str]]: 1 ctx1j

502 """Further break down components by splitting at relational and logical operators""" 

503 pattern = "|".join([f"\\b{operator.value}\\b" for operator in LogicalOperator]) 9 ctx1gihfebcda

504 logical_operators = re.compile(f"({pattern})", flags=re.IGNORECASE) 9 ctx1gihfebcda

505 

506 in_list = False 9 ctx1gihfebcda

507 base_components: list[str | list] = [] 9 ctx1gihfebcda

508 list_value_components = [] 9 ctx1gihfebcda

509 for component in components: 9 ctx1gihfebcda

510 # parse out lists as their own singular sub component 

511 subcomponents = component.split(QueryFilterBuilder.l_list_sep) 9 ctx1gihfebcda

512 for i, subcomponent in enumerate(subcomponents): 9 ctx1gihfebcda

513 if not i: 9 ctx1gihfebcda

514 continue 9 ctx1gihfebcda

515 

516 for j, list_value_string in enumerate(subcomponent.split(QueryFilterBuilder.r_list_sep)): 4 ctx1gfea

517 if j % 2: 4 ctx1gfea

518 continue (empty)

519 

520 list_value_components.append( 4 ctx1gfea

521 [val.strip() for val in list_value_string.split(QueryFilterBuilder.list_item_sep)] 

522 ) 

523 

524 quote_offset = 0 9 ctx1gihfebcda

525 subcomponents = component.split('"') 9 ctx1gihfebcda

526 for i, subcomponent in enumerate(subcomponents): 9 ctx1gihfebcda

527 # we are in a list subcomponent, which is already handled 

528 if in_list: 9 ctx1gihfebcda

529 if QueryFilterBuilder.r_list_sep in subcomponent: 4 ctx1gfea

530 # filter out the remainder of the list subcomponent and continue parsing 

531 base_components.append(list_value_components.pop(0)) (empty)

532 subcomponent = subcomponent.split(QueryFilterBuilder.r_list_sep, maxsplit=1)[-1].strip() (empty)

533 in_list = False (empty)

534 else: 

535 continue 4 ctx1gfea

536 

537 # don't parse components comprised of only a separator 

538 if subcomponent in QueryFilterBuilder.group_seps: 9 ctx1gihfebcda

539 quote_offset += 1 2 ctx1ea

540 base_components.append(subcomponent) 2 ctx1ea

541 continue 2 ctx1ea

542 

543 # this subcomponent was surrounded in quotes, so we keep it as-is 

544 if (i + quote_offset) % 2: 9 ctx1gihfebcda

545 base_components.append(f'"{subcomponent.strip()}"') 2 ctx1ea

546 continue 2 ctx1ea

547 

548 # if the final subcomponent has quotes, it creates an extra empty subcomponent at the end 

549 if not subcomponent: 9 ctx1gihfebcda

550 continue (empty)

551 

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: 9 ctx1gihfebcda

554 subcomponent, _new_sub_component = subcomponent.split(QueryFilterBuilder.l_list_sep, maxsplit=1) 4 ctx1gfea

555 subcomponent = subcomponent.strip() 4 ctx1gfea

556 subcomponents.insert(i + 1, _new_sub_component) 4 ctx1gfea

557 quote_offset += 1 4 ctx1gfea

558 in_list = True 4 ctx1gfea

559 

560 # parse out logical operators 

561 new_components = [ 9 ctx1gihfebcda

562 base_component.strip() for base_component in logical_operators.split(subcomponent) if base_component 

563 ] 

564 

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: 9 ctx1gihfebcda

568 if not component: 568 ↛ 569line 568 didn't jump to line 569 because the condition on line 568 was never true9 ctx1gihfebcda

569 continue 

570 

571 # we try relational operators first since they aren't required to be surrounded by spaces 

572 parsed_component = RelationalOperator.parse_component(component) 9 ctx1gihfebcda

573 if parsed_component is not None: 9 ctx1gihfebcda

574 base_components.extend(parsed_component) 8 ctx1ghfebcda

575 continue 8 ctx1ghfebcda

576 

577 parsed_component = RelationalKeyword.parse_component(component) 9 ctx1gihfebcda

578 if parsed_component is not None: 578 ↛ 579line 578 didn't jump to line 579 because the condition on line 578 was never true9 ctx1gihfebcda

579 base_components.extend(parsed_component) 

580 continue 

581 

582 # this component does not have any keywords or operators, so we just add it as-is 

583 base_components.append(component) 9 ctx1gihfebcda

584 

585 return base_components 9 ctx1gihfebcda

586 

587 @staticmethod 1 ctx1j

588 def _parse_base_components_into_filter_components( 1 ctx1j

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] 9 ctx1gihfebcda

593 relational_operators = [op.value for op in RelationalOperator] 9 ctx1gihfebcda

594 logical_operators = [op.value for op in LogicalOperator] 9 ctx1gihfebcda

595 

596 # parse QueryFilterComponents and logical operators 

597 components: list[str | QueryFilterBuilderComponent | LogicalOperator] = [] 9 ctx1gihfebcda

598 for i, base_component in enumerate(base_components): 9 ctx1gihfebcda

599 if isinstance(base_component, list): 9 ctx1gihfebcda

600 continue (empty)

601 

602 if base_component in QueryFilterBuilder.group_seps: 9 ctx1gihfebcda

603 components.append(base_component) (empty)

604 

605 elif base_component in relational_keywords or base_component in relational_operators: 9 ctx1gihfebcda

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 true8 ctx1ghfebcda

608 relationship = RelationalKeyword(base_components[i]) 

609 else: 

610 relationship = RelationalOperator(base_components[i]) 8 ctx1ghfebcda

611 

612 components.append( 8 ctx1ghfebcda

613 QueryFilterBuilderComponent( 

614 attribute_name=base_components[i - 1], # type: ignore 

615 relationship=relationship, 

616 value=base_components[i + 1], 

617 ) 

618 ) 

619 

620 elif base_component.upper() in logical_operators: 9 ctx1gihfebcda

621 components.append(LogicalOperator(base_component.upper())) 4 ctx1bcda

622 

623 return components 9 ctx1gihfebcda

624 

625 def as_json_model(self) -> QueryFilterJSON: 1 ctx1j

626 parts: list[QueryFilterJSONPart] = [] (empty)

627 

628 current_part: QueryFilterJSONPart | None = None (empty)

629 left_parens: list[str] = [] (empty)

630 right_parens: list[str] = [] (empty)

631 last_logical_operator: LogicalOperator | None = None (empty)

632 

633 def add_part(): (empty)

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

636 return (empty)

637 

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 

641 

642 parts.append(current_part) 

643 current_part = None 

644 left_parens.clear() 

645 right_parens.clear() 

646 last_logical_operator = None 

647 

648 for component in self.filter_components: 648 ↛ 649line 648 didn't jump to line 649 because the loop on line 648 never started(empty)

649 if isinstance(component, QueryFilterBuilderComponent): 

650 if current_part: 

651 add_part() 

652 current_part = component.as_json_model() 

653 

654 elif isinstance(component, LogicalOperator): 

655 if current_part: 

656 add_part() 

657 last_logical_operator = component 

658 

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) 

664 

665 # add last part, if any 

666 add_part() (empty)

667 return QueryFilterJSON(parts=parts) (empty)