Skip to content

clause

FilterGroup

Filter groups are used in complex queries condition to group and and or clauses in where condition

Source code in ormar/queryset/clause.py
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
class FilterGroup:
    """
    Filter groups are used in complex queries condition to group and and or
    clauses in where condition
    """

    def __init__(
        self,
        *args: Any,
        _filter_type: FilterType = FilterType.AND,
        _exclude: bool = False,
        **kwargs: Any,
    ) -> None:
        self.filter_type = _filter_type
        self.exclude = _exclude
        self._nested_groups: List["FilterGroup"] = list(args)
        self._resolved = False
        self.is_source_model_filter = False
        self._kwargs_dict = kwargs
        self.actions: List[FilterAction] = []

    def __and__(self, other: "FilterGroup") -> "FilterGroup":
        return FilterGroup(self, other)

    def __or__(self, other: "FilterGroup") -> "FilterGroup":
        return FilterGroup(self, other, _filter_type=FilterType.OR)

    def __invert__(self) -> "FilterGroup":
        self.exclude = not self.exclude
        return self

    def resolve(
        self,
        model_cls: Type["Model"],
        select_related: List = None,
        filter_clauses: List = None,
    ) -> Tuple[List[FilterAction], List[str]]:
        """
        Resolves the FilterGroups actions to use proper target model, replace
        complex relation prefixes if needed and nested groups also resolved.

        :param model_cls: model from which the query is run
        :type model_cls: Type["Model"]
        :param select_related: list of models to join
        :type select_related: List[str]
        :param filter_clauses: list of filter conditions
        :type filter_clauses: List[FilterAction]
        :return: list of filter conditions and select_related list
        :rtype: Tuple[List[FilterAction], List[str]]
        """
        select_related = select_related if select_related is not None else []
        filter_clauses = filter_clauses if filter_clauses is not None else []
        qryclause = QueryClause(
            model_cls=model_cls,
            select_related=select_related,
            filter_clauses=filter_clauses,
        )
        own_filter_clauses, select_related = qryclause.prepare_filter(
            _own_only=True, **self._kwargs_dict
        )
        self.actions = own_filter_clauses
        filter_clauses = filter_clauses + own_filter_clauses
        self._resolved = True
        if self._nested_groups:
            for group in self._nested_groups:
                (filter_clauses, select_related) = group.resolve(
                    model_cls=model_cls,
                    select_related=select_related,
                    filter_clauses=filter_clauses,
                )
        return filter_clauses, select_related

    def _iter(self) -> Generator:
        """
        Iterates all actions in a tree
        :return: generator yielding from own actions and nested groups
        :rtype: Generator
        """
        for group in self._nested_groups:
            yield from group._iter()
        yield from self.actions

    def _get_text_clauses(self) -> List[sqlalchemy.sql.expression.TextClause]:
        """
        Helper to return list of text queries from actions and nested groups
        :return: list of text queries from actions and nested groups
        :rtype: List[sqlalchemy.sql.elements.TextClause]
        """
        return [x.get_text_clause() for x in self._nested_groups] + [
            x.get_text_clause() for x in self.actions
        ]

    def get_text_clause(self) -> sqlalchemy.sql.expression.TextClause:
        """
        Returns all own actions and nested groups conditions compiled and joined
        inside parentheses.
        Escapes characters if it's required.
        Substitutes values of the models if value is a ormar Model with its pk value.
        Compiles the clause.

        :return: complied and escaped clause
        :rtype: sqlalchemy.sql.elements.TextClause
        """
        if self.filter_type == FilterType.AND:
            clause = sqlalchemy.sql.and_(*self._get_text_clauses()).self_group()
        else:
            clause = sqlalchemy.sql.or_(*self._get_text_clauses()).self_group()
        if self.exclude:
            clause = sqlalchemy.sql.not_(clause)
        return clause

get_text_clause()

Returns all own actions and nested groups conditions compiled and joined inside parentheses. Escapes characters if it's required. Substitutes values of the models if value is a ormar Model with its pk value. Compiles the clause.

Returns:

Type Description
sqlalchemy.sql.elements.TextClause

complied and escaped clause

Source code in ormar/queryset/clause.py
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
def get_text_clause(self) -> sqlalchemy.sql.expression.TextClause:
    """
    Returns all own actions and nested groups conditions compiled and joined
    inside parentheses.
    Escapes characters if it's required.
    Substitutes values of the models if value is a ormar Model with its pk value.
    Compiles the clause.

    :return: complied and escaped clause
    :rtype: sqlalchemy.sql.elements.TextClause
    """
    if self.filter_type == FilterType.AND:
        clause = sqlalchemy.sql.and_(*self._get_text_clauses()).self_group()
    else:
        clause = sqlalchemy.sql.or_(*self._get_text_clauses()).self_group()
    if self.exclude:
        clause = sqlalchemy.sql.not_(clause)
    return clause

resolve(model_cls, select_related=None, filter_clauses=None)

Resolves the FilterGroups actions to use proper target model, replace complex relation prefixes if needed and nested groups also resolved.

Parameters:

Name Type Description Default
model_cls Type['Model']

model from which the query is run

required
select_related List

list of models to join

None
filter_clauses List

list of filter conditions

None

Returns:

Type Description
Tuple[List[FilterAction], List[str]]

list of filter conditions and select_related list

Source code in ormar/queryset/clause.py
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
def resolve(
    self,
    model_cls: Type["Model"],
    select_related: List = None,
    filter_clauses: List = None,
) -> Tuple[List[FilterAction], List[str]]:
    """
    Resolves the FilterGroups actions to use proper target model, replace
    complex relation prefixes if needed and nested groups also resolved.

    :param model_cls: model from which the query is run
    :type model_cls: Type["Model"]
    :param select_related: list of models to join
    :type select_related: List[str]
    :param filter_clauses: list of filter conditions
    :type filter_clauses: List[FilterAction]
    :return: list of filter conditions and select_related list
    :rtype: Tuple[List[FilterAction], List[str]]
    """
    select_related = select_related if select_related is not None else []
    filter_clauses = filter_clauses if filter_clauses is not None else []
    qryclause = QueryClause(
        model_cls=model_cls,
        select_related=select_related,
        filter_clauses=filter_clauses,
    )
    own_filter_clauses, select_related = qryclause.prepare_filter(
        _own_only=True, **self._kwargs_dict
    )
    self.actions = own_filter_clauses
    filter_clauses = filter_clauses + own_filter_clauses
    self._resolved = True
    if self._nested_groups:
        for group in self._nested_groups:
            (filter_clauses, select_related) = group.resolve(
                model_cls=model_cls,
                select_related=select_related,
                filter_clauses=filter_clauses,
            )
    return filter_clauses, select_related

QueryClause

Constructs FilterActions from strings passed as arguments

Source code in ormar/queryset/clause.py
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
class QueryClause:
    """
    Constructs FilterActions from strings passed as arguments
    """

    def __init__(
        self, model_cls: Type["Model"], filter_clauses: List, select_related: List
    ) -> None:

        self._select_related = select_related[:]
        self.filter_clauses = filter_clauses[:]

        self.model_cls = model_cls
        self.table = self.model_cls.Meta.table

    def prepare_filter(  # noqa: A003
        self, _own_only: bool = False, **kwargs: Any
    ) -> Tuple[List[FilterAction], List[str]]:
        """
        Main external access point that processes the clauses into sqlalchemy text
        clauses and updates select_related list with implicit related tables
        mentioned in select_related strings but not included in select_related.

        :param _own_only:
        :type _own_only:
        :param kwargs: key, value pair with column names and values
        :type kwargs: Any
        :return: Tuple with list of where clauses and updated select_related list
        :rtype: Tuple[List[sqlalchemy.sql.elements.TextClause], List[str]]
        """
        if kwargs.get("pk"):
            pk_name = self.model_cls.get_column_alias(self.model_cls.Meta.pkname)
            kwargs[pk_name] = kwargs.pop("pk")

        filter_clauses, select_related = self._populate_filter_clauses(
            _own_only=_own_only, **kwargs
        )

        return filter_clauses, select_related

    def _populate_filter_clauses(
        self, _own_only: bool, **kwargs: Any
    ) -> Tuple[List[FilterAction], List[str]]:
        """
        Iterates all clauses and extracts used operator and field from related
        models if needed. Based on the chain of related names the target table
        is determined and the final clause is escaped if needed and compiled.

        :param kwargs: key, value pair with column names and values
        :type kwargs: Any
        :return: Tuple with list of where clauses and updated select_related list
        :rtype: Tuple[List[sqlalchemy.sql.elements.TextClause], List[str]]
        """
        filter_clauses = self.filter_clauses
        own_filter_clauses = []
        select_related = list(self._select_related)

        for key, value in kwargs.items():
            filter_action = FilterAction(
                filter_str=key, value=value, model_cls=self.model_cls
            )
            select_related = filter_action.update_select_related(
                select_related=select_related
            )

            own_filter_clauses.append(filter_action)

        self._register_complex_duplicates(select_related)
        filter_clauses = self._switch_filter_action_prefixes(
            filter_clauses=filter_clauses + own_filter_clauses
        )
        if _own_only:
            return own_filter_clauses, select_related
        return filter_clauses, select_related

    def _register_complex_duplicates(self, select_related: List[str]) -> None:
        """
        Checks if duplicate aliases are presented which can happen in self relation
        or when two joins end with the same pair of models.

        If there are duplicates, the all duplicated joins are registered as source
        model and whole relation key (not just last relation name).

        :param select_related: list of relation strings
        :type select_related: List[str]
        :return: None
        :rtype: None
        """
        prefixes = self._parse_related_prefixes(select_related=select_related)

        manager = self.model_cls.Meta.alias_manager
        filtered_prefixes = sorted(prefixes, key=lambda x: x.table_prefix)
        grouped = itertools.groupby(filtered_prefixes, key=lambda x: x.table_prefix)
        for _, group in grouped:
            sorted_group = sorted(
                group, key=lambda x: len(x.relation_str), reverse=True
            )
            for prefix in sorted_group[:-1]:
                if prefix.alias_key not in manager:
                    manager.add_alias(alias_key=prefix.alias_key)

    def _parse_related_prefixes(self, select_related: List[str]) -> List[Prefix]:
        """
        Walks all relation strings and parses the target models and prefixes.

        :param select_related: list of relation strings
        :type select_related: List[str]
        :return: list of parsed prefixes
        :rtype: List[Prefix]
        """
        prefixes: List[Prefix] = []
        for related in select_related:
            prefix = Prefix(
                self.model_cls,
                *get_relationship_alias_model_and_str(
                    self.model_cls, related.split("__")
                ),
            )
            prefixes.append(prefix)
        return prefixes

    def _switch_filter_action_prefixes(
        self, filter_clauses: List[FilterAction]
    ) -> List[FilterAction]:
        """
        Substitutes aliases for filter action if the complex key (whole relation str) is
        present in alias_manager.

        :param filter_clauses: raw list of actions
        :type filter_clauses: List[FilterAction]
        :return: list of actions with aliases changed if needed
        :rtype: List[FilterAction]
        """

        for action in filter_clauses:
            if isinstance(action, FilterGroup):
                for action2 in action._iter():
                    self._verify_prefix_and_switch(action2)
            else:
                self._verify_prefix_and_switch(action)
        return filter_clauses

    def _verify_prefix_and_switch(self, action: "FilterAction") -> None:
        """
        Helper to switch prefix to complex relation one if required
        :param action: action to switch prefix in
        :type action: ormar.queryset.actions.filter_action.FilterAction
        """
        manager = self.model_cls.Meta.alias_manager
        new_alias = manager.resolve_relation_alias(self.model_cls, action.related_str)
        if "__" in action.related_str and new_alias:
            action.table_prefix = new_alias

prepare_filter(_own_only=False, **kwargs)

Main external access point that processes the clauses into sqlalchemy text clauses and updates select_related list with implicit related tables mentioned in select_related strings but not included in select_related.

Parameters:

Name Type Description Default
_own_only bool False
kwargs Any

key, value pair with column names and values

required

Returns:

Type Description
Tuple[List[sqlalchemy.sql.elements.TextClause], List[str]]

Tuple with list of where clauses and updated select_related list

Source code in ormar/queryset/clause.py
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
def prepare_filter(  # noqa: A003
    self, _own_only: bool = False, **kwargs: Any
) -> Tuple[List[FilterAction], List[str]]:
    """
    Main external access point that processes the clauses into sqlalchemy text
    clauses and updates select_related list with implicit related tables
    mentioned in select_related strings but not included in select_related.

    :param _own_only:
    :type _own_only:
    :param kwargs: key, value pair with column names and values
    :type kwargs: Any
    :return: Tuple with list of where clauses and updated select_related list
    :rtype: Tuple[List[sqlalchemy.sql.elements.TextClause], List[str]]
    """
    if kwargs.get("pk"):
        pk_name = self.model_cls.get_column_alias(self.model_cls.Meta.pkname)
        kwargs[pk_name] = kwargs.pop("pk")

    filter_clauses, select_related = self._populate_filter_clauses(
        _own_only=_own_only, **kwargs
    )

    return filter_clauses, select_related

and_(*args, **kwargs)

Construct and filter from nested groups and keyword arguments

Parameters:

Name Type Description Default
args Tuple[FilterGroup]

nested filter groups

required
kwargs Any

fields names and proper value types

required

Returns:

Type Description
ormar.queryset.clause.FilterGroup

FilterGroup ready to be resolved

Source code in ormar/queryset/clause.py
147
148
149
150
151
152
153
154
155
156
157
158
def and_(*args: FilterGroup, **kwargs: Any) -> FilterGroup:
    """
    Construct and filter from nested groups and keyword arguments

    :param args: nested filter groups
    :type args: Tuple[FilterGroup]
    :param kwargs: fields names and proper value types
    :type kwargs: Any
    :return: FilterGroup ready to be resolved
    :rtype: ormar.queryset.clause.FilterGroup
    """
    return FilterGroup(_filter_type=FilterType.AND, *args, **kwargs)

or_(*args, **kwargs)

Construct or filter from nested groups and keyword arguments

Parameters:

Name Type Description Default
args Tuple[FilterGroup]

nested filter groups

required
kwargs Any

fields names and proper value types

required

Returns:

Type Description
ormar.queryset.clause.FilterGroup

FilterGroup ready to be resolved

Source code in ormar/queryset/clause.py
133
134
135
136
137
138
139
140
141
142
143
144
def or_(*args: FilterGroup, **kwargs: Any) -> FilterGroup:
    """
    Construct or filter from nested groups and keyword arguments

    :param args: nested filter groups
    :type args: Tuple[FilterGroup]
    :param kwargs: fields names and proper value types
    :type kwargs: Any
    :return: FilterGroup ready to be resolved
    :rtype: ormar.queryset.clause.FilterGroup
    """
    return FilterGroup(_filter_type=FilterType.OR, *args, **kwargs)