Coverage for src/mafw/db/db_filter.py: 99%
508 statements
« prev ^ index » next coverage.py v7.15.0, created at 2026-07-26 09:13 +0000
1# Copyright 2025–2026 European Union
2# Author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
3# SPDX-License-Identifier: EUPL-1.2
4"""
5Database filter module for MAFW.
7This module provides classes and utilities for creating and managing database filters
8using Peewee ORM. It supports various filtering operations including simple conditions,
9logical combinations, and conditional filters where one field's criteria depend on another.
11The module implements a flexible filter system that can handle:
12 - Simple field comparisons (equality, inequality, greater/less than, etc.)
13 - Complex logical operations (AND, OR, NOT)
14 - Conditional filters with dependent criteria
15 - Nested logical expressions
16 - Support for various data types and operations
18Key components include:
19 - :class:`FilterNode`: Abstract base class for filter nodes
20 - :class:`ConditionNode`: Represents individual field conditions
21 - :class:`LogicalNode`: Combines filter nodes with logical operators
22 - :class:`ConditionalNode`: Wraps conditional filter conditions
23 - :class:`ModelFilter`: Main class for building and applying filters to models
24 - :class:`ProcessorFilter`: Container for multiple model filters in a processor
26The module uses a hierarchical approach to build filter expressions that can be converted
27into Peewee expressions for database queries. It supports both simple and complex filtering
28scenarios through a combination of direct field conditions and logical expressions.
30.. versionchanged:: v2.0.0
31 Major overhaul introducing conditional filters and logical expression support.
33Example usage::
35 from mafw.db.db_filter import ModelFilter
37 # Create a simple filter
38 flt = ModelFilter(
39 'Processor.__filter__.Model',
40 field1='value1',
41 field2={'op': 'IN', 'value': [1, 2, 3]},
42 )
44 # Bind to a model and generate query
45 flt.bind(MyModel)
46 query = MyModel.select().where(flt.filter())
48.. seealso::
50 :link:`peewee` - The underlying ORM library used for database operations
52 :class:`~.mafw.enumerators.LogicalOp` - Logical operation enumerations used in filters
53"""
55import logging
56import operator
57import re
58from collections import OrderedDict, UserDict
59from collections.abc import Iterable, Sequence
60from copy import copy
61from functools import reduce
62from typing import TYPE_CHECKING, Any, Literal, Self, TypeAlias, cast
64import peewee
65from peewee import Model
67from mafw.db.db_model import mafw_model_register
68from mafw.enumerators import LogicalOp
70log = logging.getLogger(__name__)
73Token = tuple[str, str]
74"""Type definition for a logical expression token"""
77def _format_expected(expected: str | Sequence[str]) -> str:
78 """Format a description of expected tokens for diagnostics."""
79 if isinstance(expected, str): 79 ↛ 81line 79 didn't jump to line 81 because the condition on line 79 was always true
80 return expected
81 return ' or '.join(expected)
84# 1. An atom is a tuple of the literal string 'NAME' and the value
85NameNode = tuple[Literal['NAME'], str]
86"""An atom is a tuple of the literal string 'NAME' and the value"""
88# 2. A NOT node is a tuple of 'NOT' and a recursive node
89# We use a string forward reference 'ExprNode' because it is defined below
90NotNode = tuple[Literal['NOT'], 'ExprNode']
91"""A NOT node is a tuple of 'NOT' and a recursive node"""
93# 3. AND/OR nodes are tuples of the operator and two recursive nodes
94BinaryNode = tuple[Literal['AND', 'OR'], 'ExprNode', 'ExprNode']
95"""AND/OR nodes are tuples of the operator and two recursive nodes"""
97# 4. The main recursive type combining all options
98ExprNode: TypeAlias = NameNode | NotNode | BinaryNode
99"""
100The main recursive type combining all options
102This type represents the abstract syntax tree (AST) nodes used in logical expressions.
103It can be one of:
105 - :data:`NameNode`: A named element (field name or filter name)
106 - :data:`NotNode`: A negation operation
107 - :data:`BinaryNode`: An AND/OR operation between two nodes
108"""
110TOKEN_SPECIFICATION = [
111 ('LPAREN', r'\('),
112 ('RPAREN', r'\)'),
113 ('AND', r'\bAND\b'),
114 ('OR', r'\bOR\b'),
115 ('NOT', r'\bNOT\b'),
116 ('NAME', r'[A-Za-z_][A-Za-z0-9_\.]*(?:\:[A-Za-z_][A-Za-z0-9_]*)?'),
117 ('SKIP', r'[ \t\n\r]+'),
118 ('MISMATCH', r'.'),
119]
120"""Token specifications"""
122MASTER_RE = re.compile('|'.join(f'(?P<{name}>{pattern})' for name, pattern in TOKEN_SPECIFICATION))
123"""Compiled regular expression to interpret the logical expression grammar"""
126class ParseError(ValueError):
127 """Base exception for logical expression parsing failures."""
129 def __init__(self, message: str, *, position: int | None = None) -> None:
130 super().__init__(message)
131 self.position = position
134class UnexpectedTokenError(ParseError):
135 """Raised when a token is present but not valid in the current context."""
137 def __init__(
138 self,
139 token: Token,
140 *,
141 expected: str | Sequence[str] | None = None,
142 position: int | None = None,
143 ) -> None:
144 expected_desc = f'; expected {_format_expected(expected)}' if expected else ''
145 message = f'Unexpected token {token[0]} ({token[1]}) at position {position}{expected_desc}'
146 super().__init__(message, position=position)
147 self.token = token
148 self.expected = expected
151class UnexpectedEndOfExpressionError(ParseError):
152 """Raised when the expression ends before the parser could finish."""
154 def __init__(self, *, expected: str | Sequence[str] | None = None, position: int | None = None) -> None:
155 expected_desc = f'; expected {_format_expected(expected)}' if expected else ''
156 message = f'Unexpected end of expression at position {position}{expected_desc}'
157 super().__init__(message, position=position)
158 self.expected = expected
161class MissingTokenError(ParseError):
162 """Raised when a specific token was required but missing."""
164 def __init__(self, expected: str | Sequence[str], *, position: int | None = None) -> None:
165 message = f'Expected {_format_expected(expected)} before end of expression at position {position}'
166 super().__init__(message, position=position)
167 self.expected = expected
170class UnknownNameError(ParseError):
171 """Raised when a NAME token is not in the supplied whitelist."""
173 def __init__(
174 self,
175 name: str,
176 *,
177 valid_names: Iterable[str],
178 position: int | None = None,
179 ) -> None:
180 valid_list = sorted(valid_names)
181 allowed = ', '.join(valid_list[:5])
182 if len(valid_list) > 5: 182 ↛ 183line 182 didn't jump to line 183 because the condition on line 182 was never true
183 allowed = allowed + ', ...'
184 message = f'Unknown name {name!r} at position {position}; valid names: {allowed}'
185 super().__init__(message, position=position)
186 self.name = name
187 self.valid_names = tuple(valid_list)
190def _tokenize_with_positions(text: str) -> tuple[list[Token], list[int]]:
191 """Tokenize text while capturing the start offset of each token."""
192 tokens: list[Token] = []
193 positions: list[int] = []
194 for mo in MASTER_RE.finditer(text):
195 kind = mo.lastgroup
196 value = mo.group()
197 if kind == 'SKIP':
198 continue
199 elif kind == 'MISMATCH':
200 raise ParseError(f'Unexpected character {value!r}', position=mo.start())
201 else:
202 assert kind is not None
203 tokens.append((kind, value))
204 positions.append(mo.start())
205 return tokens, positions
208def tokenize(text: str) -> list[Token]:
209 """
210 Tokenize a logical expression string into a list of tokens.
212 This function breaks down a logical expression string into individual
213 tokens based on the defined token specifications. It skips whitespace
214 and raises a :exc:`ParseError` for unexpected characters.
216 :param text: The logical expression string to tokenize
217 :type text: str
218 :return: A list of tokens represented as (token_type, token_value) tuples
219 :rtype: list[:data:`Token`]
220 :raises ParseError: If an unexpected character is encountered in the text
221 """
222 tokens, _ = _tokenize_with_positions(text)
223 return tokens
226class ExprParser:
227 """
228 Recursive descent parser producing a simple Abstract Syntax Tree (AST).
230 The parser handles logical expressions with the following grammar:
232 .. code-block:: none
234 expr := or_expr
235 or_expr := and_expr ("OR" and_expr)*
236 and_expr:= not_expr ("AND" not_expr)*
237 not_expr:= "NOT" not_expr | atom
238 atom := NAME | "(" expr ")"
240 AST nodes are tuples representing different constructs:
242 - ("NAME", "token"): A named element (field name or filter name)
243 - ("NOT", node): A negation operation
244 - ("AND", left, right): An AND operation between two nodes
245 - ("OR", left, right): An OR operation between two nodes
247 .. versionadded:: v2.0.0
249 To help users diagnose grammar and semantic problems, the parser now
250 reports detailed error classes with character offsets and accepts
251 an optional ``valid_names`` iterable to reject unknown identifiers early.
252 """
254 def __init__(self, text: str, *, valid_names: Iterable[str] | None = None) -> None:
255 """
256 Initialize the expression parser with a logical expression string.
258 :param text: The logical expression to parse
259 :type text: str
260 :param valid_names: Optional whitelist of valid NAME tokens
261 :type valid_names: Iterable[str] | None
262 """
263 self._text = text
264 self.tokens, self.token_positions = _tokenize_with_positions(text)
265 self.pos = 0
266 self._valid_names = frozenset(valid_names) if valid_names is not None else None
268 def peek(self) -> Token | None:
269 """
270 Peek at the next token without consuming it.
272 :return: The next token if available, otherwise None
273 :rtype: :data:`Token` | None
274 """
275 if self.pos < len(self.tokens):
276 return self.tokens[self.pos]
277 return None
279 def accept(self, *kinds: str) -> Token | None:
280 """
281 Accept and consume the next token if it matches one of the given types.
283 :param kinds: Token types to accept
284 :type kinds: str
285 :return: The consumed token if matched, otherwise None
286 :rtype: :data:`Token` | None
287 """
288 tok = self.peek()
289 if tok and tok[0] in kinds:
290 self.pos += 1
291 return tok
292 return None
294 def _current_position(self) -> int:
295 """Return the character offset of the next token or the end of input."""
296 if self.pos < len(self.token_positions):
297 return self.token_positions[self.pos]
298 return len(self._text)
300 def expect(self, kind: str) -> 'Token':
301 """
302 Expect and consume a specific token type.
304 :param kind: The expected token type
305 :type kind: str
306 :return: The consumed token
307 :rtype: :data:`Token`
308 :raises ParseError: If the expected token is not found
309 """
310 tok = self.accept(kind)
311 if tok:
312 return tok
313 position = self._current_position()
314 current = self.peek()
315 if not current:
316 raise MissingTokenError(kind, position=position)
317 raise UnexpectedTokenError(current, expected=kind, position=position)
319 def parse(self) -> 'ExprNode':
320 """
321 Parse the entire logical expression and return the resulting AST.
323 :return: The abstract syntax tree representation of the expression
324 :rtype: :data:`ExprNode`
325 :raises ParseError: If the expression is malformed
326 """
327 node = self.parse_or()
328 if self.pos != len(self.tokens):
329 token = self.tokens[self.pos]
330 position = self.token_positions[self.pos]
331 raise UnexpectedTokenError(token, expected='end of expression', position=position)
332 return node
334 def parse_or(self) -> 'ExprNode':
335 """
336 Parse an OR expression.
338 :return: The parsed OR expression AST node
339 :rtype: :data:`ExprNode`
340 """
341 left = self.parse_and()
342 while self.accept('OR'):
343 right = self.parse_and()
344 left = ('OR', left, right)
345 return left
347 def parse_and(self) -> 'ExprNode':
348 """
349 Parse an AND expression.
351 :return: The parsed AND expression AST node
352 :rtype: :data:`ExprNode`
353 """
354 left = self.parse_not()
355 while self.accept('AND'):
356 right = self.parse_not()
357 left = ('AND', left, right)
358 return left
360 def parse_not(self) -> 'ExprNode':
361 """
362 Parse a NOT expression.
364 :return: The parsed NOT expression AST node
365 :rtype: :data:`ExprNode`
366 """
367 if self.accept('NOT'):
368 node = self.parse_not()
369 return 'NOT', node
370 return self.parse_atom()
372 def parse_atom(self) -> 'ExprNode':
373 """
374 Parse an atomic expression (NAME or parenthesised expression).
376 :return: The parsed atomic expression AST node
377 :rtype: :data:`ExprNode`
378 :raises ParseError: If an unexpected token is encountered
379 """
380 tok = self.peek()
381 if not tok:
382 raise UnexpectedEndOfExpressionError(position=self._current_position())
383 if tok[0] == 'LPAREN':
384 self.accept('LPAREN')
385 node = self.parse_or()
386 self.expect('RPAREN')
387 return node
388 elif tok[0] == 'NAME':
389 start_pos = self.token_positions[self.pos]
390 self.accept('NAME')
391 name = tok[1]
392 valid_names = self._valid_names
393 if valid_names is not None and name not in valid_names:
394 raise UnknownNameError(name, valid_names=valid_names, position=start_pos)
395 return 'NAME', name
396 else:
397 raise UnexpectedTokenError(tok, expected='NAME or LPAREN', position=self._current_position())
400def ast_to_string(ast: ExprNode) -> str:
401 """
402 Convert an abstract syntax tree (AST) back to its string representation.
404 :param ast: The AST to convert
405 :type ast: ExprNode
406 :return: The string representation of the AST
407 :rtype: str
408 """
409 t = ast[0]
410 if t == 'NAME':
411 return cast(NameNode, ast)[1]
412 elif t == 'NOT':
413 inner = cast(NotNode, ast)[1]
414 inner_str = ast_to_string(inner)
415 if inner[0] in ('AND', 'OR'):
416 return f'NOT ({inner_str})'
417 return f'NOT {inner_str}'
418 elif t == 'AND':
419 bin_ast = cast(BinaryNode, ast)
420 left, right = bin_ast[1], bin_ast[2]
421 left_str = ast_to_string(left)
422 right_str = ast_to_string(right)
423 if left[0] == 'OR':
424 left_str = f'({left_str})'
425 if right[0] == 'OR':
426 right_str = f'({right_str})'
427 return f'{left_str} AND {right_str}'
428 elif t == 'OR': 428 ↛ 434line 428 didn't jump to line 434 because the condition on line 428 was always true
429 bin_ast = cast(BinaryNode, ast)
430 left_str = ast_to_string(bin_ast[1])
431 right_str = ast_to_string(bin_ast[2])
432 return f'{left_str} OR {right_str}'
433 else:
434 raise ValueError(f'Unsupported AST node type: {t}')
437class FilterNode:
438 """Abstract base for nodes."""
440 def to_expression(self, model: type[Model]) -> peewee.Expression | bool:
441 raise NotImplementedError # pragma: no cover
444class ConditionNode(FilterNode):
445 """
446 Represents a single condition node in a filter expression.
448 This class encapsulates a single filtering condition that can be applied
449 to a model field. It supports various logical operations through the
450 :class:`.LogicalOp` enumerator or string representations of operations.
452 .. versionadded:: v2.0.0
453 """
455 def __init__(self, field: str | None, operation: LogicalOp | str, value: Any, name: str | None = None):
456 """
457 Initialize a condition node.
459 :param field: The name of the field to apply the condition to.
460 :type field: str | None
461 :param operation: The logical operation to perform.
462 :type operation: LogicalOp | str
463 :param value: The value to compare against.
464 :type value: Any
465 :param name: Optional name for this condition node.
466 :type name: str | None, Optional
467 """
468 self.field = field # may be None for some special nodes
469 if isinstance(operation, str):
470 try:
471 self.operation = LogicalOp(operation)
472 except ValueError:
473 raise ValueError(f'Unsupported operation: {operation}')
474 else:
475 self.operation = operation
476 self.value = value
477 self.name = name
479 def to_expression(self, model: type[Model]) -> peewee.Expression:
480 """
481 Convert this condition node to a Peewee expression.
483 This method translates the condition represented by this node into
484 a Peewee expression that can be used in database queries.
486 :param model: The model class containing the field to filter.
487 :type model: type[Model]
488 :return: A Peewee expression representing this condition.
489 :rtype: peewee.Expression
490 :raises RuntimeError: If the node has no field to evaluate.
491 :raises ValueError: If an unsupported operation is specified.
492 :raises TypeError: If operation requirements are not met (e.g., IN operation requires list/tuple).
493 """
494 if self.field is None:
495 # Should not happen for standard ConditionNode
496 raise RuntimeError('ConditionNode has no field to evaluate')
497 model_field = getattr(model, self.field)
498 op = self.operation
499 val = self.value
500 # the code is full of cast and redundant checks to make mypy happy.
501 # I do not know to which extent they make the code safer, but for sure they make it less readable.
502 if op == LogicalOp.EQ:
503 return cast(peewee.Expression, cast(object, model_field == val))
504 elif op == LogicalOp.NE:
505 return cast(peewee.Expression, cast(object, model_field != val))
506 elif op == LogicalOp.LT:
507 return cast(peewee.Expression, cast(object, model_field < val))
508 elif op == LogicalOp.LE:
509 return cast(peewee.Expression, cast(object, model_field <= val))
510 elif op == LogicalOp.GT:
511 return cast(peewee.Expression, cast(object, model_field > val))
512 elif op == LogicalOp.GE:
513 return cast(peewee.Expression, cast(object, model_field >= val))
514 elif op == LogicalOp.GLOB:
515 return cast(peewee.Expression, model_field % val)
516 elif op == LogicalOp.LIKE:
517 return cast(peewee.Expression, model_field**val)
518 elif op == LogicalOp.REGEXP:
519 if hasattr(model_field, 'regexp') and callable(getattr(model_field, 'regexp')):
520 return cast(peewee.Expression, getattr(model_field, 'regexp')(val))
521 else:
522 raise ValueError(f'REGEXP operation not supported for field type {type(model_field)}')
523 elif op == LogicalOp.IN:
524 if not isinstance(val, (list, tuple)):
525 raise TypeError(f'IN operation requires list/tuple, got {type(val)}')
526 if hasattr(model_field, 'in_') and callable(getattr(model_field, 'in_')):
527 return cast(peewee.Expression, getattr(model_field, 'in_')(val))
528 else:
529 raise ValueError(f'IN operation not supported for field type {type(model_field)}')
530 elif op == LogicalOp.NOT_IN:
531 if not isinstance(val, (list, tuple)):
532 raise TypeError(f'NOT_IN operation requires list/tuple, got {type(val)}')
533 if hasattr(model_field, 'not_in') and callable(getattr(model_field, 'not_in')):
534 return cast(peewee.Expression, getattr(model_field, 'not_in')(val))
535 else:
536 raise ValueError(f'NOT_IN operation not supported for field type {type(model_field)}')
537 elif op == LogicalOp.BETWEEN:
538 if not isinstance(val, (list, tuple)) or len(val) != 2:
539 raise TypeError(f'BETWEEN operation requires list/tuple of 2 elements, got {val}')
540 if hasattr(model_field, 'between') and callable(getattr(model_field, 'between')):
541 return cast(peewee.Expression, getattr(model_field, 'between')(val[0], val[1]))
542 else:
543 raise ValueError(f'BETWEEN operation not supported for field type {type(model_field)}')
544 elif op == LogicalOp.BIT_AND:
545 if hasattr(model_field, 'bin_and') and callable(getattr(model_field, 'bin_and')):
546 return cast(peewee.Expression, cast(object, getattr(model_field, 'bin_and')(val) != 0))
547 else:
548 raise ValueError(f'BIT_AND operation not supported for field type {type(model_field)}')
549 elif op == LogicalOp.BIT_OR:
550 if hasattr(model_field, 'bin_or') and callable(getattr(model_field, 'bin_or')):
551 return cast(peewee.Expression, cast(object, getattr(model_field, 'bin_or')(val) != 0))
552 else:
553 raise ValueError(f'BIT_OR operation not supported for field type {type(model_field)}')
554 elif op == LogicalOp.IS_NULL:
555 return cast(peewee.Expression, model_field.is_null())
556 elif op == LogicalOp.IS_NOT_NULL:
557 return cast(peewee.Expression, model_field.is_null(False))
558 else:
559 raise ValueError(f'Unsupported operation: {op}')
562class ConditionalNode(FilterNode):
563 """
564 Wraps :class:`ConditionalFilterCondition` behaviour as a :class:`FilterNode`.
566 This class serves as an adapter to integrate conditional filter conditions
567 into the filter node hierarchy, allowing them to be treated uniformly with
568 other filter nodes during expression evaluation.
570 .. versionadded:: v2.0.0
571 """
573 def __init__(self, conditional: 'ConditionalFilterCondition', name: str | None = None):
574 """
575 Initialize a conditional node.
577 :param conditional: The conditional filter condition to wrap
578 :type conditional: ConditionalFilterCondition
579 :param name: Optional name for this conditional node
580 :type name: str | None, Optional
581 """
582 self.conditional = conditional
583 self.name = name
585 def to_expression(self, model: type[Model]) -> peewee.Expression:
586 """
587 Convert this conditional node to a Peewee expression.
589 This method delegates the conversion to the wrapped conditional filter
590 condition's :meth:`to_expression` method.
592 :param model: The model class to generate the expression for
593 :type model: type[Model]
594 :return: A Peewee expression representing this conditional node
595 :rtype: peewee.Expression
596 """
597 return self.conditional.to_expression(model)
600class LogicalNode(FilterNode):
601 """
602 Logical combination of child nodes.
604 This class represents logical operations (AND, OR, NOT) applied to filter nodes.
605 It enables building complex filter expressions by combining simpler filter nodes
606 with logical operators.
608 .. versionadded:: v2.0.0
609 """
611 def __init__(self, op: str, *children: FilterNode):
612 """
613 Initialize a logical node.
615 :param op: The logical operation ('AND', 'OR', 'NOT')
616 :type op: str
617 :param children: Child filter nodes to combine with the logical operation
618 :type children: FilterNode
619 """
620 self.op = op # 'AND', 'OR', 'NOT'
621 self.children = list(children)
623 def to_expression(self, model: type[Model]) -> peewee.Expression | bool:
624 """
625 Convert this logical node to a Peewee expression.
627 This method evaluates the logical operation on the child nodes and returns
628 the corresponding Peewee expression.
630 :param model: The model class to generate the expression for
631 :type model: type[Model]
632 :return: A Peewee expression representing this logical node
633 :rtype: peewee.Expression | bool
634 :raises ValueError: If an unknown logical operation is specified
635 """
636 if self.op == 'NOT':
637 assert len(self.children) == 1
638 inner = self.children[0].to_expression(model)
639 return cast(peewee.Expression, ~inner)
640 elif self.op == 'AND':
641 expressions = [c.to_expression(model) for c in self.children]
642 return cast(peewee.Expression, reduce(operator.and_, expressions))
643 elif self.op == 'OR':
644 expressions = [c.to_expression(model) for c in self.children]
645 return cast(peewee.Expression, reduce(operator.or_, expressions))
646 else:
647 raise ValueError(f'Unknown logical op: {self.op}')
650class ConditionalFilterCondition:
651 """
652 Represents a conditional filter where one field's criteria depends on another.
654 This allows expressing logic like:
655 "IF field_a IN [x, y] THEN field_b IN [1, 2] ELSE no constraint on field_b"
657 Example usage:
659 .. code-block:: python
661 # Filter: sample_id in [1,2] if composite_image_id in [100,101]
662 condition = ConditionalFilterCondition(
663 condition_field='composite_image_id',
664 condition_op='IN',
665 condition_value=[100, 101],
666 then_field='sample_id',
667 then_op='IN',
668 then_value=[1, 2],
669 )
671 # This generates:
672 # WHERE (composite_image_id IN (100, 101) AND sample_id IN (1, 2))
673 # OR (composite_image_id NOT IN (100, 101))
674 """
676 def __init__(
677 self,
678 condition_field: str,
679 condition_op: str | LogicalOp,
680 condition_value: Any,
681 then_field: str,
682 then_op: str | LogicalOp,
683 then_value: Any,
684 else_field: str | None = None,
685 else_op: str | LogicalOp | None = None,
686 else_value: Any | None = None,
687 name: str | None = None,
688 ) -> None:
689 """
690 Initialise a conditional filter condition.
692 :param condition_field: The field to check for the condition
693 :type condition_field: str
694 :param condition_op: The operation for the condition (e.g., 'IN', '==')
695 :type condition_op: str | LogicalOp
696 :param condition_value: The value(s) for the condition
697 :type condition_value: Any
698 :param then_field: The field to filter when condition is true
699 :type then_field: str
700 :param then_op: The operation to apply when condition is true
701 :type then_op: str | LogicalOp
702 :param then_value: The value(s) for the then clause
703 :type then_value: Any
704 :param else_field: Optional field to filter when condition is false
705 :type else_field: str | None
706 :param else_op: Optional operation when condition is false
707 :type else_op: str | LogicalOp | None
708 :param else_value: Optional value(s) for the else clause
709 :type else_value: Any | None
710 :param name: The name of this condition. Avoid name clashing with model fields. Defaults to None
711 :type name: str | None, Optional
712 """
713 self.condition_field = condition_field
714 self.condition_op = condition_op
715 self.condition_value = condition_value
716 self.then_field = then_field
717 self.then_op = then_op
718 self.then_value = then_value
719 self.else_field = else_field
720 self.else_op = else_op
721 self.else_value = else_value
722 self.name = name
724 def to_expression(self, model: type[Model]) -> peewee.Expression:
725 """
726 Convert this conditional filter to a Peewee expression.
728 The resulting expression is:
729 (condition AND then_constraint) OR (NOT condition AND else_constraint)
731 Which logically means:
733 - When condition is true, apply then_constraint
734 - When condition is false, apply else_constraint (or no constraint)
736 :param model: The model class containing the fields
737 :type model: type[Model]
738 :return: A Peewee expression
739 :rtype: peewee.Expression
740 """
741 # Build the condition expression
742 condition_expr = ConditionNode(self.condition_field, self.condition_op, self.condition_value).to_expression(
743 model
744 )
746 # Build the then expression
747 then_expr = ConditionNode(self.then_field, self.then_op, self.then_value).to_expression(model)
749 # Build the else expression
750 if self.else_field is not None and self.else_op is not None:
751 else_expr = ConditionNode(self.else_field, self.else_op, self.else_value).to_expression(model)
752 else:
753 # No constraint in else clause - always true
754 # the nested cast is needed to make mypy happy.
755 else_expr = cast(peewee.Expression, cast(object, True))
757 # Combine: (condition AND then) OR (NOT condition AND else)
758 return cast(peewee.Expression, (condition_expr & then_expr) | (~condition_expr & else_expr))
760 def __eq__(self, other: Any) -> bool:
761 if not isinstance(other, ConditionalFilterCondition):
762 return False
764 return vars(self) == vars(other)
767class ModelFilter:
768 r"""
769 Class to filter rows from a model.
771 The filter object can be used to generate a where clause to be applied to Model.select().
773 The construction of a ModelFilter is normally done via a configuration file using the :meth:`from_conf` class method.
774 The name of the filter is playing a key role in this. If it follows a dot structure like:
776 *ProcessorName.__filter__.ModelName*
778 then the corresponding table from the TOML configuration object will be used.
780 For each processor, there might be many Filters, up to one for each Model used to get the input list. If a
781 processor is joining together three Models when performing the input select, there will be up to three Filters
782 collaborating on making the selection.
784 The filter configuration can contain the following key, value pair:
786 - key / string pairs, where the key is the name of a field in the corresponding Model
788 - key / numeric pairs
790 - key / arrays
792 - key / dict pairs with 'op' and 'value' keys for explicit operation specification
794 All fields from the configuration file will be added to the instance namespace, thus accessible with the dot
795 notation. Moreover, the field names and their filter value will be added to a private dictionary to simplify the
796 generation of the filter SQL code.
798 The user can use the filter object to store selection criteria. He can construct queries using the filter
799 contents in the same way as he could use processor parameters.
801 If he wants to automatically generate valid filtering expression, he can use the :meth:`filter` method. In order
802 for this to work, the ModelFilter object be :meth:`bound <bind>` to a Model. Without this binding the ModelFilter will not
803 be able to automatically generate expressions.
805 For each field in the filter, one condition will be generated according to the following scheme:
807 ================= ================= ==================
808 Filter field type Logical operation Example
809 ================= ================= ==================
810 Numeric, boolean == Field == 3.14
811 String GLOB Field GLOB '\*ree'
812 List IN Field IN [1, 2, 3]
813 Dict (explicit) op from dict Field BIT_AND 5
814 ================= ================= ==================
816 All conditions will be joined with a AND logic by default, but this can be changed.
818 The ModelFilter also supports logical expressions to combine multiple filter conditions using AND, OR, and NOT
819 operators. These expressions can reference named filter conditions within the same filter or even combine
820 conditions from different filters when used with :class:`ProcessorFilter`.
822 Conditional filters allow expressing logic like:
823 "IF field_a IN [x, y] THEN field_b IN [1, 2] ELSE no constraint on field_b"
825 Consider the following example:
827 .. code-block:: python
828 :linenos:
830 class MeasModel(MAFwBaseModel):
831 meas_id = AutoField(primary_key=True)
832 sample_name = TextField()
833 successful = BooleanField()
834 flags = IntegerField()
835 composite_image_id = IntegerField()
836 sample_id = IntegerField()
839 # Traditional simplified usage
840 flt = ModelFilter(
841 'MyProcessor.__filter__.MyModel',
842 sample_name='sample_00*',
843 meas_id=[1, 2, 3],
844 successful=True,
845 )
847 # New explicit operation usage
848 flt = ModelFilter(
849 'MyProcessor.__filter__.MyModel',
850 sample_name={'op': 'LIKE', 'value': 'sample_00%'},
851 flags={'op': 'BIT_AND', 'value': 5},
852 meas_id={'op': 'IN', 'value': [1, 2, 3]},
853 )
855 # Logical expression usage
856 flt = ModelFilter(
857 'MyProcessor.__filter__.MyModel',
858 sample_name={'op': 'LIKE', 'value': 'sample_00%'},
859 flags={'op': 'BIT_AND', 'value': 5},
860 meas_id={'op': 'IN', 'value': [1, 2, 3]},
861 __logic__='sample_name AND (flags OR meas_id)',
862 )
864 # Conditional filter usage
865 flt = ModelFilter(
866 'MyProcessor.__filter__.MyModel',
867 sample_name='sample_00*',
868 composite_image_id=[100, 101],
869 sample_id=[1, 2],
870 __conditional__=[
871 {
872 'condition_field': 'composite_image_id',
873 'condition_op': 'IN',
874 'condition_value': [100, 101],
875 'then_field': 'sample_id',
876 'then_op': 'IN',
877 'then_value': [1, 2],
878 }
879 ],
880 )
882 flt.bind(MeasModel)
883 filtered_query = MeasModel.select().where(flt.filter())
885 The explicit operation format allows for bitwise operations and other advanced filtering.
887 TOML Configuration Examples:
889 .. code-block:: toml
891 [MyProcessor.__filter__.MyModel]
892 sample_name = "sample_00*" # Traditional GLOB
893 successful = true # Traditional equality
895 # Explicit operations
896 flags = { op = "BIT_AND", value = 5 }
897 score = { op = ">=", value = 75.0 }
898 category = { op = "IN", value = ["A", "B", "C"] }
899 date_range = { op = "BETWEEN", value = ["2024-01-01", "2024-12-31"] }
901 # Logical expression for combining conditions
902 __logic__ = "sample_name AND (successful OR flags)"
904 # Conditional filters
905 [[MyProcessor.__filter__.MyModel.__conditional__]]
906 condition_field = "composite_image_id"
907 condition_op = "IN"
908 condition_value = [100, 101]
909 then_field = "sample_id"
910 then_op = "IN"
911 then_value = [1, 2]
913 # Nested conditions with logical expressions
914 [MyProcessor.__filter__.MyModel.nested_conditions]
915 __logic__ = "a OR b"
916 a = { op = "LIKE", value = "test%" }
917 b = { op = "IN", value = [1, 2, 3] }
919 .. seealso::
921 - :class:`mafw.db.db_filter.ProcessorFilter` - For combining multiple ModelFilters with logical expressions
922 - :class:`mafw.db.db_filter.ConditionalFilterCondition` - For conditional filtering logic
923 - :class:`mafw.db.db_filter.ExprParser` - For parsing logical expressions
924 """
926 logic_name = '__logic__'
927 """
928 The logic keyword identifier.
930 This value cannot be used as field name in the filter bound model.
931 """
932 conditional_name = '__conditional__'
933 """
934 The conditional keyword identifier.
936 This value cannot be used as field name in the filter bound model.
937 """
939 def __init__(self, name_: str, **kwargs: Any) -> None:
940 """
941 Constructor parameters:
943 :param `name_`: The name of the filter. It should be in dotted format to facilitate the configuration via the
944 steering file. The _ is used to allow the user to have a keyword argument named name.
945 :type `name_`: str
946 :param kwargs: Keyword parameters corresponding to fields and filter values.
948 .. versionchanged:: v1.2.0
949 The parameter *name* has been renamed as *name_*.
951 .. versionchanged:: v1.3.0
952 Implementation of explicit operation.
954 .. versionchanged:: v2.0.0
955 Introduction of conditional filters, logical expression and hierarchical structure.
956 Introduction of autobinding for MAFwBaseModels
958 """
959 self.name = name_
960 self.model_name = name_.split('.')[-1]
961 self.model: type[Model] | None = None
962 self._model_bound = False
964 # attempt to autobind
965 self._auto_bind()
967 # mapping name -> FilterNode
968 self._nodes: 'OrderedDict[str, FilterNode]' = OrderedDict()
969 # conditional nodes mapping (named)
970 self._cond_nodes: 'OrderedDict[str, ConditionalNode]' = OrderedDict()
971 # logic expression for this filter (combining top-level node names)
972 self._logic_expr: str | None = None
974 # Extract conditional filters if present
975 if self.conditional_name in kwargs:
976 conditionals = kwargs.pop(self.conditional_name)
977 if not isinstance(conditionals, list):
978 conditionals = [conditionals]
980 for cond_dict in conditionals:
981 self.add_conditional_from_dict(cond_dict)
983 # Extract logic for internal conditions, if provided
984 if self.logic_name in kwargs:
985 self._logic_expr = kwargs.pop(self.logic_name)
987 # now process remaining kwargs as either:
988 # - simple/extended condition for a field
989 # - or a nested mapping describing subconditions for field (field-level logic)
990 for k, v in kwargs.items():
991 # simple types map to ConditionNode
992 if isinstance(v, dict) and ('op' in v and 'value' in v):
993 # explicit op/value for field k
994 # extended operation condition
995 node = ConditionNode(k, v['op'], v['value'], name=k)
996 self._nodes[k] = node
997 elif isinstance(v, dict) and any(
998 isinstance(x, dict) or x == self.logic_name or x not in ['op', 'value']
999 for x in v.keys()
1000 if isinstance(v, dict)
1001 ):
1002 # nested mapping: create sub-nodes for this field
1003 # v expected like {'__logic__': 'a OR b', 'a': {'op':..., 'value':...}, 'b': ...}
1004 subnodes: 'OrderedDict[str, FilterNode]' = OrderedDict()
1005 sub_logic = v.get(self.logic_name, None)
1006 for subk, subv in v.items():
1007 if subk == self.logic_name:
1008 continue
1009 if isinstance(subv, dict) and ('op' in subv and 'value' in subv):
1010 subnode = ConditionNode(k, subv['op'], subv['value'], name=subk)
1011 subnodes[subk] = subnode
1012 else:
1013 subnodes[subk] = self._create_condition_node_from_value(subv, k, subk)
1014 # combine subnodes using sub_logic or AND by default
1015 if sub_logic:
1016 ast = ExprParser(sub_logic).parse()
1017 ln = self._build_logical_node_from_ast(ast, subnodes, model_name_placeholder=k)
1018 else:
1019 # AND all subnodes
1020 ln = LogicalNode('AND', *subnodes.values())
1021 self._nodes[k] = ln
1022 else:
1023 self._nodes[k] = self._create_condition_node_from_value(v, k, k)
1025 def _auto_bind(self) -> None:
1026 """
1027 Attempt to automatically bind the filter to a model.
1029 This method tries to retrieve the model associated with the filter's model name from the
1030 :mod:`mafw.db.db_model` registry and bind it using the :meth:`bind` method.
1032 If the model cannot be found,a warning is logged indicating the failure to perform auto-binding.
1034 This model is automatically invoked by the :class:`.ModelFilter` constructor.
1035 """
1036 try:
1037 model = mafw_model_register.get_model(self.model_name)
1038 self.bind(model) # type: ignore[arg-type]
1039 except KeyError:
1040 log.warning(f'Impossible to perform auto-binding for model {self.model_name}')
1042 def _build_logical_node_from_ast(
1043 self, ast: ExprNode, name_to_nodes: dict[str, FilterNode], model_name_placeholder: str | None = None
1044 ) -> FilterNode:
1045 """Recursively build LogicalNode from AST using a mapping name->FilterNode."""
1046 t = ast[0]
1047 if t == 'NAME':
1048 named_ast = cast(NameNode, ast)
1049 nm = named_ast[1]
1050 if nm not in name_to_nodes:
1051 raise KeyError(f'Unknown name {nm} in nested logic for field {model_name_placeholder}')
1052 return name_to_nodes[nm]
1053 elif t == 'NOT':
1054 not_ast = cast(NotNode, ast)
1055 child = self._build_logical_node_from_ast(not_ast[1], name_to_nodes, model_name_placeholder)
1056 return LogicalNode('NOT', child)
1057 elif t in ('AND', 'OR'):
1058 bin_ast = cast(BinaryNode, ast)
1059 left = self._build_logical_node_from_ast(bin_ast[1], name_to_nodes, model_name_placeholder)
1060 right = self._build_logical_node_from_ast(bin_ast[2], name_to_nodes, model_name_placeholder)
1061 return LogicalNode(t, left, right)
1062 else:
1063 raise ValueError(f'Unsupported AST node {t}')
1065 @staticmethod
1066 def _create_condition_node_from_value(value: Any, field_name: str, node_name: str | None = None) -> ConditionNode:
1067 """
1068 Create a FilterCondition based on value type (backward compatibility).
1070 :param value: The filter value
1071 :param field_name: The field name
1072 :return: A FilterCondition
1073 """
1074 if isinstance(value, (int, float, bool)):
1075 return ConditionNode(field_name, LogicalOp.EQ, value, node_name)
1076 elif isinstance(value, str):
1077 return ConditionNode(field_name, LogicalOp.GLOB, value, node_name)
1078 elif isinstance(value, list):
1079 return ConditionNode(field_name, LogicalOp.IN, value, node_name)
1080 else:
1081 raise TypeError(f'ModelFilter value of unsupported type {type(value)} for field {field_name}.')
1083 def bind(self, model: type[Model]) -> None:
1084 """
1085 Connects a filter to a Model class.
1087 :param model: Model to be bound.
1088 :type model: Model
1089 """
1091 self.model = model
1092 self._model_bound = True
1094 if hasattr(self.model, self.logic_name) and self._model_bound:
1095 if TYPE_CHECKING:
1096 assert self.model is not None
1098 log.warning(
1099 f'Model {self.model.__name__} has a field named {self.logic_name}. This is '
1100 f'preventing the logic expression to work.'
1101 )
1102 log.warning('Modify your model. Logic expression disabled.')
1103 self._logic_expr = None
1105 @property
1106 def is_bound(self) -> bool:
1107 """Returns true if the ModelFilter has been bound to a Model"""
1108 return self._model_bound
1110 def add_conditional(self, conditional: ConditionalFilterCondition) -> None:
1111 """
1112 Add a conditional filter.
1114 .. versionadded:: v2.0.0
1116 :param conditional: The conditional filter condition
1117 :type conditional: ConditionalFilterCondition
1118 """
1119 condition_name = conditional.name
1120 if condition_name is None:
1121 # it means that the user did not specify any name for this condition.
1122 # we will then assign one
1123 increment = 0
1124 while True:
1125 condition_name = f'__cond{increment + len(self._cond_nodes)}__'
1126 if condition_name not in self._cond_nodes:
1127 break
1128 else:
1129 increment += 1
1130 else:
1131 # the user specified a name for this condition. we will use it but first we check if it is not yet used
1132 if condition_name in self._cond_nodes:
1133 raise KeyError(
1134 f'A conditional filter named {condition_name} already exists. Please review your steering file.'
1135 )
1137 node = ConditionalNode(conditional, name=condition_name)
1138 self._cond_nodes[condition_name] = node
1139 self._nodes[condition_name] = node
1141 def add_conditional_from_dict(self, config: dict[str, Any]) -> None:
1142 """
1143 Add a conditional filter from a configuration dictionary.
1145 .. versionadded:: v2.0.0
1147 :param config: Dictionary with conditional filter configuration
1148 :type config: dict[str, Any]
1149 """
1150 conditional = ConditionalFilterCondition(
1151 condition_field=config['condition_field'],
1152 condition_op=config['condition_op'],
1153 condition_value=config['condition_value'],
1154 then_field=config['then_field'],
1155 then_op=config['then_op'],
1156 then_value=config['then_value'],
1157 else_field=config.get('else_field'),
1158 else_op=config.get('else_op'),
1159 else_value=config.get('else_value'),
1160 name=config.get('name'),
1161 )
1162 self.add_conditional(conditional)
1164 @classmethod
1165 def from_conf(cls, name: str, conf: dict[str, Any]) -> Self:
1166 """
1167 Builds a Filter object from a steering file dictionary.
1169 If the name is in dotted notation, then this should be corresponding to the table in the configuration file.
1170 If a default configuration is provided, this will be used as a starting point for the filter, and it will be
1171 updated by the actual configuration in ``conf``.
1173 In normal use, you would provide the specific configuration via the conf parameter.
1175 See details in the :class:`class documentation <ModelFilter>`
1177 :param name: The name of the filter in dotted notation.
1178 :type name: str
1179 :param conf: The configuration dictionary.
1180 :type conf: dict
1181 :return: A Filter object
1182 :rtype: ModelFilter
1183 """
1184 param = {}
1186 # split the name from dotted notation
1187 # ProcessorName#123.ModelName.Filter
1188 # the processor name is actually the processor replica name
1189 names = name.split('.')
1190 if len(names) == 3 and names[1] == '__filter__':
1191 proc_name, _, model_name = names
1192 if proc_name in conf and '__filter__' in conf[proc_name] and model_name in conf[proc_name]['__filter__']:
1193 param.update(copy(conf[proc_name]['__filter__'][model_name]))
1195 # if the name is not in the expected dotted notation, the use an empty filter.
1196 return cls(name, **param)
1198 def _evaluate_logic_ast(self, ast: ExprNode) -> peewee.Expression | bool:
1199 """
1200 Evaluate an abstract syntax tree (AST) representing a logical expression.
1202 This method recursively evaluates the AST nodes to produce a Peewee expression
1203 or boolean value representing the logical combination of filter conditions.
1205 :param ast: The abstract syntax tree node to evaluate
1206 :type ast: Any
1207 :return: A Peewee expression for logical operations or boolean True/False
1208 :rtype: peewee.Expression | bool
1209 :raises KeyError: If a referenced condition name is not found in the filter
1210 :raises ValueError: If an unsupported AST node type is encountered
1211 """
1212 t = ast[0]
1213 if t == 'NAME':
1214 named_ast = cast(NameNode, ast)
1215 nm = named_ast[1]
1216 if nm not in self._nodes:
1217 raise KeyError(f"Unknown node '{nm}' in logic for filter {self.name}")
1218 node = self._nodes[nm]
1220 if TYPE_CHECKING:
1221 assert self.model is not None
1222 return node.to_expression(self.model)
1223 elif t == 'NOT':
1224 not_ast = cast(NotNode, ast)
1225 val = self._evaluate_logic_ast(not_ast[1])
1226 return cast(peewee.Expression, ~val)
1227 elif t == 'AND':
1228 bin_ast = cast(BinaryNode, ast)
1229 left = self._evaluate_logic_ast(bin_ast[1])
1230 right = self._evaluate_logic_ast(bin_ast[2])
1231 return cast(peewee.Expression, cast(object, left & right))
1232 elif t == 'OR':
1233 bin_ast = cast(BinaryNode, ast)
1234 left = self._evaluate_logic_ast(bin_ast[1])
1235 right = self._evaluate_logic_ast(bin_ast[2])
1236 return cast(peewee.Expression, cast(object, left | right))
1237 else:
1238 raise ValueError(f'Unsupported AST node {t}')
1240 def filter(self, join_with: Literal['AND', 'OR'] = 'AND') -> peewee.Expression | bool:
1241 """
1242 Generates a filtering expression joining all filtering fields.
1244 See details in the :class:`class documentation <ModelFilter>`
1246 .. versionchanged:: v1.3.0
1247 Add the possibility to specify a `join_with` function
1249 .. versionchanged:: v2.0.0
1250 Add support for conditional filters and for logical expression
1252 :param join_with: How to join conditions ('AND' or 'OR'). Defaults to 'AND'.
1253 :type join_with: Literal['AND', 'OR'], default 'AND'
1254 :return: The filtering expression.
1255 :rtype: peewee.Expression | bool
1256 :raises TypeError: when the field value type is not supported.
1257 :raises ValueError: when join_with is not 'AND' or 'OR'.
1258 """
1259 if not self.is_bound:
1260 log.warning('Unable to generate the filter. Did you bind the filter to the model?')
1261 return True
1263 if TYPE_CHECKING:
1264 # if we get here, it means that we have a valid model
1265 assert self.model is not None
1267 # if logic provided for this filter, use it
1268 if self._logic_expr:
1269 try:
1270 ast = ExprParser(self._logic_expr).parse()
1271 except ParseError as e:
1272 raise ValueError(f'Error parsing logic for filter {self.name}: {e}')
1273 try:
1274 return self._evaluate_logic_ast(ast)
1275 except KeyError as e:
1276 raise ValueError(f'Error evaluating logic for filter {self.name}: {e}')
1278 # otherwise combine all top-level nodes (AND by default)
1279 exprs = [n.to_expression(self.model) for n in self._nodes.values()]
1280 if not exprs:
1281 return True
1282 if join_with not in ('AND', 'OR'):
1283 raise ValueError("join_with must be 'AND' or 'OR'")
1284 if join_with == 'AND':
1285 return cast(peewee.Expression, reduce(operator.and_, exprs))
1286 return cast(peewee.Expression, reduce(operator.or_, exprs))
1289class ProcessorFilter(UserDict[str, ModelFilter]):
1290 """
1291 A special dictionary to store all :class:`Filters <mafw.db.db_filter.ModelFilter>` in a processors.
1293 It contains a publicly accessible dictionary with the configuration of each ModelFilter using the Model name as
1294 keyword.
1296 It contains a private dictionary with the global filter configuration as well.
1297 The global filter is not directly accessible, but only some of its members will be exposed via properties.
1298 In particular, the new_only flag that is relevant only at the Processor level can be accessed directly using the
1299 :attr:`new_only`. If not specified in the configuration file, the new_only is by default True.
1301 It is possible to assign a logic operation string to the register that is used to join all the filters together
1302 when performing the :meth:`filter_all`. If no logic operation string is provided, the register will provide a join
1303 condition using either AND (default) or OR.
1304 """
1306 def __init__(self, data: dict[str, ModelFilter] | None = None, /, **kwargs: Any) -> None:
1307 """
1308 Constructor parameters:
1310 :param data: Initial data
1311 :type data: dict
1312 :param kwargs: Keywords arguments
1313 """
1314 self._global_filter: dict[str, Any] = {}
1315 self._logic: str | None = None
1316 super().__init__(data, **kwargs)
1318 @property
1319 def new_only(self) -> bool:
1320 """
1321 The new only flag.
1323 :return: True, if only new items, not already in the output database table must be processed.
1324 :rtype: bool
1325 """
1326 return cast(bool, self._global_filter.get('new_only', True))
1328 @new_only.setter
1329 def new_only(self, v: bool) -> None:
1330 self._global_filter['new_only'] = v
1332 def __setitem__(self, key: str, value: ModelFilter) -> None:
1333 """
1334 Set a new value at key.
1336 If value is not a Filter, then it will be automatically and silently discarded.
1338 :param key: Dictionary key. Normally the name of the model linked to the filter.
1339 :type key: str
1340 :param value: The Filter.
1341 :type value: ModelFilter
1342 """
1343 if not isinstance(value, ModelFilter):
1344 return
1345 super().__setitem__(key, value)
1347 def bind_all(self, models: list[type[Model]] | dict[str, type[Model]]) -> None:
1348 """
1349 Binds all filters to their models.
1351 The ``models`` list or dictionary should contain a valid model for all the ModelFilters in the registry.
1352 In the case of a dictionary, the key value should be the model name.
1354 :param models: List or dictionary of a databank of Models from which the ModelFilter can be bound.
1355 :type models: list[type(Model)] | dict[str,type(Model)]
1356 """
1357 if isinstance(models, list):
1358 models = {m.__name__: m for m in models}
1360 # check, if we have a filter for each listed models, if not create one using the default configuration.
1361 for model_name in models.keys():
1362 if model_name not in self.data:
1363 self.data[model_name] = ModelFilter.from_conf(f'{model_name}', conf={})
1365 for k, v in self.data.items():
1366 if k in self.data and k in models and not v.is_bound: 1366 ↛ 1365line 1366 didn't jump to line 1365 because the condition on line 1366 was always true
1367 v.bind(models[k])
1369 def filter_all(self, join_with: Literal['AND', 'OR'] = 'AND') -> peewee.Expression | bool:
1370 """
1371 Generates a where clause joining all filters.
1373 If a logic expression is present, it will be used to combine named filters.
1374 Otherwise, fall back to the legacy behaviour using join_with.
1376 :raise ValueError: If the parsing of the logical expression fails
1377 :param join_with: Logical function to join the filters if no logic expression is provided.
1378 :type join_with: Literal['AND', 'OR'], default: 'AND'
1379 :return: ModelFilter expression
1380 :rtype: peewee.Expression
1381 """
1382 # If a logic expression is present at the global level, use it to combine filters
1383 if self._logic:
1384 try:
1385 ast = ExprParser(self._logic).parse()
1386 except ParseError as e:
1387 raise ValueError(f'Error parsing global logic for ProcessorFilter: {e}')
1389 def eval_ast(node: ExprNode) -> peewee.Expression | bool:
1390 t = node[0]
1391 if t == 'NAME':
1392 named_node = cast(NameNode, node)
1393 nm = named_node[1]
1394 if nm not in self.data:
1395 raise KeyError(f"Unknown filter name '{nm}' in processor logic")
1396 flt = self.data[nm]
1397 if not flt.is_bound:
1398 log.warning(f"ModelFilter '{nm}' is not bound; using True for its expression")
1399 return True
1400 return flt.filter()
1401 elif t == 'NOT':
1402 not_node = cast(NotNode, node)
1403 return cast(peewee.Expression, ~eval_ast(not_node[1]))
1404 elif t == 'AND':
1405 bin_node = cast(BinaryNode, node)
1406 return cast(peewee.Expression, cast(object, eval_ast(bin_node[1]) & eval_ast(bin_node[2])))
1407 elif t == 'OR':
1408 bin_node = cast(BinaryNode, node)
1409 return cast(peewee.Expression, cast(object, eval_ast(bin_node[1]) | eval_ast(bin_node[2])))
1410 else:
1411 raise ValueError(f'Unsupported AST node {t}')
1413 try:
1414 return eval_ast(ast)
1415 except KeyError as e:
1416 raise ValueError(f'Error evaluating processor logic: {e}')
1418 # Legacy behaviour: combine all filters with join_with (AND/OR)
1419 filter_list = [flt.filter() for flt in self.data.values() if flt.is_bound]
1420 if join_with == 'AND':
1421 return cast(peewee.Expression, cast(object, reduce(operator.and_, filter_list, True)))
1422 else:
1423 return cast(peewee.Expression, cast(object, reduce(operator.or_, filter_list, True)))