Skip to content

query_action

QueryAction

Bases: abc.ABC

Base QueryAction class with common params for Filter and Order actions.

Source code in ormar\queryset\actions\query_action.py
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
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
class QueryAction(abc.ABC):
    """
    Base QueryAction class with common params for Filter and Order actions.
    """

    def __init__(self, query_str: str, model_cls: Type["Model"]) -> None:
        self.query_str = query_str
        self.field_name: str = ""
        self.related_parts: List[str] = []
        self.related_str: str = ""

        self.table_prefix = ""
        self.source_model = model_cls
        self.target_model = model_cls
        self.is_through = False

        self._split_value_into_parts(query_str)
        self._determine_filter_target_table()

    def __eq__(self, other: object) -> bool:  # pragma: no cover
        if not isinstance(other, QueryAction):
            return False
        return self.query_str == other.query_str

    def __hash__(self) -> Any:
        return hash((self.table_prefix, self.query_str))

    @abc.abstractmethod
    def _split_value_into_parts(self, query_str: str) -> None:  # pragma: no cover
        """
        Splits string into related parts and field_name
        :param query_str: query action string to split (i..e filter or order by)
        :type query_str: str
        """
        pass

    @abc.abstractmethod
    def get_text_clause(
        self,
    ) -> sqlalchemy.sql.expression.TextClause:  # pragma: no cover
        pass

    @property
    def table(self) -> sqlalchemy.Table:
        """Shortcut to sqlalchemy Table of filtered target model"""
        return self.target_model.Meta.table

    @property
    def column(self) -> sqlalchemy.Column:
        """Shortcut to sqlalchemy column of filtered target model"""
        aliased_name = self.target_model.get_column_alias(self.field_name)
        return self.target_model.Meta.table.columns[aliased_name]

    def update_select_related(self, select_related: List[str]) -> List[str]:
        """
        Updates list of select related with related part included in the filter key.
        That way If you want to just filter by relation you do not have to provide
        select_related separately.

        :param select_related: list of relation join strings
        :type select_related: List[str]
        :return: list of relation joins with implied joins from filter added
        :rtype: List[str]
        """
        select_related = select_related[:]
        if self.related_str and not any(
            rel.startswith(self.related_str) for rel in select_related
        ):
            select_related.append(self.related_str)
        return select_related

    def _determine_filter_target_table(self) -> None:
        """
        Walks the relation to retrieve the actual model on which the clause should be
        constructed, extracts alias based on last relation leading to target model.
        """
        (
            self.table_prefix,
            self.target_model,
            self.related_str,
            self.is_through,
        ) = get_relationship_alias_model_and_str(self.source_model, self.related_parts)

column: sqlalchemy.Column property

Shortcut to sqlalchemy column of filtered target model

table: sqlalchemy.Table property

Shortcut to sqlalchemy Table of filtered target model

Updates list of select related with related part included in the filter key. That way If you want to just filter by relation you do not have to provide select_related separately.

Parameters:

Name Type Description Default
select_related List[str]

list of relation join strings

required

Returns:

Type Description
List[str]

list of relation joins with implied joins from filter added

Source code in ormar\queryset\actions\query_action.py
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
def update_select_related(self, select_related: List[str]) -> List[str]:
    """
    Updates list of select related with related part included in the filter key.
    That way If you want to just filter by relation you do not have to provide
    select_related separately.

    :param select_related: list of relation join strings
    :type select_related: List[str]
    :return: list of relation joins with implied joins from filter added
    :rtype: List[str]
    """
    select_related = select_related[:]
    if self.related_str and not any(
        rel.startswith(self.related_str) for rel in select_related
    ):
        select_related.append(self.related_str)
    return select_related