Skip to content

actions

FilterAction

Bases: QueryAction

Filter Actions is populated by queryset when filter() is called.

All required params are extracted but kept raw until actual filter clause value is required -> then the action is converted into text() clause.

Extracted in order to easily change table prefixes on complex relations.

Source code in ormar/queryset/actions/filter_action.py
Python
 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
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
class FilterAction(QueryAction):
    """
    Filter Actions is populated by queryset when filter() is called.

    All required params are extracted but kept raw until actual filter clause value
    is required -> then the action is converted into text() clause.

    Extracted in order to easily change table prefixes on complex relations.
    """

    def __init__(self, filter_str: str, value: Any, model_cls: type["Model"]) -> None:
        super().__init__(query_str=filter_str, model_cls=model_cls)
        self.filter_value = value
        self._escape_characters_in_clause()

    def has_escaped_characters(self) -> bool:
        """Check if value is a string that contains characters to escape"""
        return isinstance(self.filter_value, str) and any(
            c for c in ESCAPE_CHARACTERS if c in self.filter_value
        )

    def _split_value_into_parts(self, query_str: str) -> None:
        parts = query_str.split("__")
        if parts[-1] in FILTER_OPERATORS:
            self.operator = parts[-1]
            self.field_name = parts[-2]
            self.related_parts = parts[:-2]
        else:
            self.operator = "exact"
            self.field_name = parts[-1]
            self.related_parts = parts[:-1]

    def _escape_characters_in_clause(self) -> None:
        """
        Escapes the special characters ["%", "_"] if needed.
        Adds `%` for `like` queries.

        :raises QueryDefinitionError: if contains or icontains is used with
        ormar model instance
        :return: escaped value and flag if escaping is needed
        :rtype: tuple[Any, bool]
        """
        self.has_escaped_character = False
        if self.operator in [
            "contains",
            "icontains",
            "startswith",
            "istartswith",
            "endswith",
            "iendswith",
        ]:
            if isinstance(self.filter_value, ormar.Model):
                raise QueryDefinitionError(
                    "You cannot use contains and icontains with instance of the Model"
                )
            self.has_escaped_character = self.has_escaped_characters()
            if self.has_escaped_character:
                self._escape_chars()
            self._prefix_suffix_quote()

    def _escape_chars(self) -> None:
        """Actually replaces chars to escape in value"""
        for char in ESCAPE_CHARACTERS:
            self.filter_value = self.filter_value.replace(char, f"\\{char}")

    def _prefix_suffix_quote(self) -> None:
        """
        Adds % to the beginning of the value if operator checks for containment and not
        starts with.

        Adds % to the end of the value if operator checks for containment and not
        end with.
        :return:
        :rtype:
        """
        prefix = "%" if "start" not in self.operator else ""
        sufix = "%" if "end" not in self.operator else ""
        self.filter_value = f"{prefix}{self.filter_value}{sufix}"

    def get_text_clause(self) -> TextClause:
        """
        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 isinstance(self.filter_value, ormar.Model):
            self.filter_value = self.filter_value.pk

        op_attr = FILTER_OPERATORS[self.operator]
        if self.operator == "isnull":
            op_attr = "is_" if self.filter_value else "isnot"
            filter_value = None
        else:
            filter_value = self.filter_value
        if self.table_prefix:
            aliased_table = (
                self.source_model.ormar_config.alias_manager.prefixed_table_name(
                    self.table_prefix, self.column.table
                )
            )
            aliased_column = getattr(aliased_table.c, self.column.name)
        else:
            aliased_column = self.column
        clause = getattr(aliased_column, op_attr)(filter_value)
        if self.has_escaped_character:
            clause.modifiers["escape"] = "\\"
        return clause

get_text_clause()

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

Source code in ormar/queryset/actions/filter_action.py
Python
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
def get_text_clause(self) -> TextClause:
    """
    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 isinstance(self.filter_value, ormar.Model):
        self.filter_value = self.filter_value.pk

    op_attr = FILTER_OPERATORS[self.operator]
    if self.operator == "isnull":
        op_attr = "is_" if self.filter_value else "isnot"
        filter_value = None
    else:
        filter_value = self.filter_value
    if self.table_prefix:
        aliased_table = (
            self.source_model.ormar_config.alias_manager.prefixed_table_name(
                self.table_prefix, self.column.table
            )
        )
        aliased_column = getattr(aliased_table.c, self.column.name)
    else:
        aliased_column = self.column
    clause = getattr(aliased_column, op_attr)(filter_value)
    if self.has_escaped_character:
        clause.modifiers["escape"] = "\\"
    return clause

has_escaped_characters()

Check if value is a string that contains characters to escape

Source code in ormar/queryset/actions/filter_action.py
Python
63
64
65
66
67
def has_escaped_characters(self) -> bool:
    """Check if value is a string that contains characters to escape"""
    return isinstance(self.filter_value, str) and any(
        c for c in ESCAPE_CHARACTERS if c in self.filter_value
    )

OrderAction

Bases: QueryAction

Order Actions is populated by queryset when order_by() is called.

All required params are extracted but kept raw until actual filter clause value is required -> then the action is converted into text() clause.

Extracted in order to easily change table prefixes on complex relations.

Source code in ormar/queryset/actions/order_action.py
Python
 13
 14
 15
 16
 17
 18
 19
 20
 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
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
class OrderAction(QueryAction):
    """
    Order Actions is populated by queryset when order_by() is called.

    All required params are extracted but kept raw until actual filter clause value
    is required -> then the action is converted into text() clause.

    Extracted in order to easily change table prefixes on complex relations.
    """

    def __init__(
        self,
        order_str: str,
        model_cls: type["Model"],
        alias: Optional[str] = None,
        nulls_ordering: Optional[str] = None,
    ) -> None:
        self.direction: str = ""
        super().__init__(query_str=order_str, model_cls=model_cls)
        self.is_source_model_order = False
        if alias:
            self.table_prefix = alias
        if self.source_model == self.target_model and "__" not in self.related_str:
            self.is_source_model_order = True
        self.nulls_ordering = nulls_ordering

    @property
    def field_alias(self) -> str:
        return self.target_model.get_column_alias(self.field_name)

    @property
    def is_postgres_bool(self) -> bool:
        dialect = self.target_model.ormar_config.database.dialect.name
        field_type = self.target_model.ormar_config.model_fields[
            self.field_name
        ].__type__
        return dialect == "postgresql" and field_type is bool

    def get_field_name_text(self) -> str:
        """
        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
        """
        prefix = f"{self.table_prefix}_" if self.table_prefix else ""
        return f"{prefix}{self.table.name}.{self.field_alias}"

    def get_min_or_max(self) -> sqlalchemy.sql.expression.TextClause:
        """
        Used in limit sub queries where you need to use aggregated functions
        in order to order by columns not included in group by. For postgres bool
        field it's using bool_or function as aggregates does not work with this type
        of columns.

        :return: min or max function to order
        :rtype: sqlalchemy.sql.elements.TextClause
        """
        reference = self.get_field_name_text()
        if self.direction == "":
            function = "min" if not self.is_postgres_bool else "bool_or"
            return text(f"{function}({reference})")
        function = "max" if not self.is_postgres_bool else "bool_or"
        return text(f"{function}({reference}) desc")

    def get_text_clause(self) -> sqlalchemy.sql.expression.TextClause:
        """
        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
        """
        dialect = self.target_model.ormar_config.database.dialect
        quoter = dialect.identifier_preparer.quote
        prefix = f"{self.table_prefix}_" if self.table_prefix else ""
        table_name = self.table.name
        field_name = self.field_alias
        if not prefix:
            table_name = quoter(table_name)
        else:
            table_name = quoter(f"{prefix}{table_name}")
        field_name = quoter(field_name)
        return text(self._build_order_expression(f"{table_name}.{field_name}"))

    def _build_order_expression(self, full_column: str) -> str:
        """
        Builds the final ORDER BY expression for a fully-qualified column,
        optionally including a `NULLS FIRST`/`NULLS LAST` annotation. On MySQL,
        which lacks the SQL:2003 `NULLS` syntax, emulate it by prepending an
        `IS NULL` / `IS NOT NULL` sort key — it only affects ordering,
        not the result set.

        :param full_column: fully-qualified, quoted column reference
        :type full_column: str
        :return: ORDER BY expression as raw SQL text
        :rtype: str
        """
        direction = f" {self.direction}" if self.direction else ""
        base = f"{full_column}{direction}"
        if self.nulls_ordering is None:
            return base
        dialect_name = self.target_model.ormar_config.database.dialect.name
        if dialect_name == "mysql":  # pragma: no cover
            not_kw = "not " if self.nulls_ordering == "first" else ""
            return f"{full_column} is {not_kw}null, {base}"
        return f"{base} nulls {self.nulls_ordering}"  # pragma: no cover

    def _split_value_into_parts(self, order_str: str) -> None:
        if order_str.startswith("-"):
            self.direction = "desc"
            order_str = order_str[1:]
        parts = order_str.split("__")
        self.field_name = parts[-1]
        self.related_parts = parts[:-1]

    @classmethod
    def from_model_defaults(cls, model_cls: type["Model"]) -> list["OrderAction"]:
        """
        Builds the default list of ``OrderAction`` instances from a model's
        ``OrmarConfig.orders_by`` (which always contains at least the primary
        key, populated by the metaclass).

        :param model_cls: model class whose defaults should be used
        :type model_cls: type["Model"]
        :return: list of default OrderAction instances for the model
        :rtype: list[OrderAction]
        """
        return [
            cls(order_str=str(name), model_cls=model_cls)
            for name in model_cls.ormar_config.orders_by
        ]

    def flipped(self) -> "OrderAction":
        """
        Returns a shallow copy of this order action with the sort direction
        and any `NULLS FIRST`/`NULLS LAST` annotation inverted.

        Used by reverse slicing to turn an ASC/DESC ordering into its mirror
        image so that ``LIMIT N`` fetches rows from the tail of the original
        ordering. Callers are responsible for reversing the result list in
        memory afterwards so the caller-visible ordering is preserved.

        :return: new OrderAction with flipped direction and nulls ordering
        :rtype: OrderAction
        """
        flipped = copy.copy(self)
        flipped.direction = "" if self.direction == "desc" else "desc"
        if self.nulls_ordering == "first":
            flipped.nulls_ordering = "last"
        elif self.nulls_ordering == "last":
            flipped.nulls_ordering = "first"
        return flipped

    def check_if_filter_apply(self, target_model: type["Model"], alias: str) -> bool:
        """
        Checks filter conditions to find if they apply to current join.

        :param target_model: model which is now processed
        :type target_model: type["Model"]
        :param alias: prefix of the relation
        :type alias: str
        :return: result of the check
        :rtype: bool
        """
        return target_model == self.target_model and alias == self.table_prefix

check_if_filter_apply(target_model, alias)

Checks filter conditions to find if they apply to current join.

:param target_model: model which is now processed :type target_model: type["Model"] :param alias: prefix of the relation :type alias: str :return: result of the check :rtype: bool

Source code in ormar/queryset/actions/order_action.py
Python
170
171
172
173
174
175
176
177
178
179
180
181
def check_if_filter_apply(self, target_model: type["Model"], alias: str) -> bool:
    """
    Checks filter conditions to find if they apply to current join.

    :param target_model: model which is now processed
    :type target_model: type["Model"]
    :param alias: prefix of the relation
    :type alias: str
    :return: result of the check
    :rtype: bool
    """
    return target_model == self.target_model and alias == self.table_prefix

flipped()

Returns a shallow copy of this order action with the sort direction and any NULLS FIRST/NULLS LAST annotation inverted.

Used by reverse slicing to turn an ASC/DESC ordering into its mirror image so that LIMIT N fetches rows from the tail of the original ordering. Callers are responsible for reversing the result list in memory afterwards so the caller-visible ordering is preserved.

:return: new OrderAction with flipped direction and nulls ordering :rtype: OrderAction

Source code in ormar/queryset/actions/order_action.py
Python
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
def flipped(self) -> "OrderAction":
    """
    Returns a shallow copy of this order action with the sort direction
    and any `NULLS FIRST`/`NULLS LAST` annotation inverted.

    Used by reverse slicing to turn an ASC/DESC ordering into its mirror
    image so that ``LIMIT N`` fetches rows from the tail of the original
    ordering. Callers are responsible for reversing the result list in
    memory afterwards so the caller-visible ordering is preserved.

    :return: new OrderAction with flipped direction and nulls ordering
    :rtype: OrderAction
    """
    flipped = copy.copy(self)
    flipped.direction = "" if self.direction == "desc" else "desc"
    if self.nulls_ordering == "first":
        flipped.nulls_ordering = "last"
    elif self.nulls_ordering == "last":
        flipped.nulls_ordering = "first"
    return flipped

from_model_defaults(model_cls) classmethod

Builds the default list of OrderAction instances from a model's OrmarConfig.orders_by (which always contains at least the primary key, populated by the metaclass).

:param model_cls: model class whose defaults should be used :type model_cls: type["Model"] :return: list of default OrderAction instances for the model :rtype: list[OrderAction]

Source code in ormar/queryset/actions/order_action.py
Python
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
@classmethod
def from_model_defaults(cls, model_cls: type["Model"]) -> list["OrderAction"]:
    """
    Builds the default list of ``OrderAction`` instances from a model's
    ``OrmarConfig.orders_by`` (which always contains at least the primary
    key, populated by the metaclass).

    :param model_cls: model class whose defaults should be used
    :type model_cls: type["Model"]
    :return: list of default OrderAction instances for the model
    :rtype: list[OrderAction]
    """
    return [
        cls(order_str=str(name), model_cls=model_cls)
        for name in model_cls.ormar_config.orders_by
    ]

get_field_name_text()

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

Source code in ormar/queryset/actions/order_action.py
Python
51
52
53
54
55
56
57
58
59
60
61
def get_field_name_text(self) -> str:
    """
    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
    """
    prefix = f"{self.table_prefix}_" if self.table_prefix else ""
    return f"{prefix}{self.table.name}.{self.field_alias}"

get_min_or_max()

Used in limit sub queries where you need to use aggregated functions in order to order by columns not included in group by. For postgres bool field it's using bool_or function as aggregates does not work with this type of columns.

:return: min or max function to order :rtype: sqlalchemy.sql.elements.TextClause

Source code in ormar/queryset/actions/order_action.py
Python
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
def get_min_or_max(self) -> sqlalchemy.sql.expression.TextClause:
    """
    Used in limit sub queries where you need to use aggregated functions
    in order to order by columns not included in group by. For postgres bool
    field it's using bool_or function as aggregates does not work with this type
    of columns.

    :return: min or max function to order
    :rtype: sqlalchemy.sql.elements.TextClause
    """
    reference = self.get_field_name_text()
    if self.direction == "":
        function = "min" if not self.is_postgres_bool else "bool_or"
        return text(f"{function}({reference})")
    function = "max" if not self.is_postgres_bool else "bool_or"
    return text(f"{function}({reference}) desc")

get_text_clause()

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

Source code in ormar/queryset/actions/order_action.py
Python
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
def get_text_clause(self) -> sqlalchemy.sql.expression.TextClause:
    """
    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
    """
    dialect = self.target_model.ormar_config.database.dialect
    quoter = dialect.identifier_preparer.quote
    prefix = f"{self.table_prefix}_" if self.table_prefix else ""
    table_name = self.table.name
    field_name = self.field_alias
    if not prefix:
        table_name = quoter(table_name)
    else:
        table_name = quoter(f"{prefix}{table_name}")
    field_name = quoter(field_name)
    return text(self._build_order_expression(f"{table_name}.{field_name}"))

SelectAction

Bases: QueryAction

Order Actions is populated by queryset when order_by() is called.

All required params are extracted but kept raw until actual filter clause value is required -> then the action is converted into text() clause.

Extracted in order to easily change table prefixes on complex relations.

Source code in ormar/queryset/actions/select_action.py
Python
12
13
14
15
16
17
18
19
20
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
class SelectAction(QueryAction):
    """
    Order Actions is populated by queryset when order_by() is called.

    All required params are extracted but kept raw until actual filter clause value
    is required -> then the action is converted into text() clause.

    Extracted in order to easily change table prefixes on complex relations.
    """

    def __init__(self, select_str: str, model_cls: type["Model"]) -> None:
        super().__init__(query_str=select_str, model_cls=model_cls)

    def _split_value_into_parts(self, order_str: str) -> None:
        parts = order_str.split("__")
        self.field_name = parts[-1]
        self.related_parts = parts[:-1]

    @property
    def is_numeric(self) -> bool:
        return self.get_target_field_type() in [int, float, decimal.Decimal]

    def get_target_field_type(self) -> Any:
        return self.target_model.ormar_config.model_fields[self.field_name].__type__

    def get_text_clause(self) -> sqlalchemy.sql.expression.ColumnClause:
        alias = f"{self.table_prefix}_" if self.table_prefix else ""
        return sqlalchemy.column(f"{alias}{self.field_name}")

    def apply_func(
        self, func: Callable, use_label: bool = True
    ) -> sqlalchemy.sql.expression.ColumnElement:
        result = func(self.get_text_clause())
        if use_label:
            rel_prefix = f"{self.related_str}__" if self.related_str else ""
            result = result.label(f"{rel_prefix}{self.field_name}")
        return result