Main entry point from outside (after proper initialization). Extracts columns list to fetch, construct all required joins for select related, then applies all conditional and sort clauses. Returns ready to run query with all joins and clauses. :ret
(self)
| 98 | ) |
| 99 | |
| 100 | def build_select_expression(self) -> sqlalchemy.sql.Select: |
| 101 | """ |
| 102 | Main entry point from outside (after proper initialization). |
| 103 | |
| 104 | Extracts columns list to fetch, |
| 105 | construct all required joins for select related, |
| 106 | then applies all conditional and sort clauses. |
| 107 | |
| 108 | Returns ready to run query with all joins and clauses. |
| 109 | |
| 110 | :return: ready to run query with all joins and clauses. |
| 111 | :rtype: sqlalchemy.sql.selectable.Select |
| 112 | """ |
| 113 | self_related_fields = self.model_cls.own_table_columns( |
| 114 | model=self.model_cls, excludable=self.excludable, use_alias=True |
| 115 | ) |
| 116 | self.columns = self.model_cls.ormar_config.alias_manager.prefixed_columns( # type: ignore |
| 117 | "", self.table, self_related_fields |
| 118 | ) |
| 119 | self.apply_order_bys_for_primary_model() |
| 120 | self.select_from = self.table |
| 121 | |
| 122 | related_models = group_related_list(self._select_related) |
| 123 | |
| 124 | for related in related_models: |
| 125 | remainder = None |
| 126 | if isinstance(related_models, dict) and related_models[related]: |
| 127 | remainder = related_models[related] |
| 128 | sql_join = SqlJoin( |
| 129 | used_aliases=self.used_aliases, |
| 130 | select_from=self.select_from, # type: ignore |
| 131 | columns=self.columns, |
| 132 | excludable=self.excludable, |
| 133 | order_columns=self.order_columns, |
| 134 | sorted_orders=self.sorted_orders, |
| 135 | main_model=self.model_cls, |
| 136 | relation_name=related, |
| 137 | relation_str=related, |
| 138 | related_models=remainder, |
| 139 | ) |
| 140 | |
| 141 | ( |
| 142 | self.used_aliases, |
| 143 | self.select_from, |
| 144 | self.columns, |
| 145 | self.sorted_orders, |
| 146 | ) = sql_join.build_join() # type: ignore |
| 147 | |
| 148 | if self._pagination_query_required(): |
| 149 | limit_qry, on_clause = self._build_pagination_condition() |
| 150 | self.select_from = sqlalchemy.sql.join( |
| 151 | cast("FromClauseRole", self.select_from), limit_qry, on_clause |
| 152 | ) |
| 153 | |
| 154 | expr = sqlalchemy.sql.select(*self.columns) |
| 155 | expr = expr.select_from(cast("FromClauseRole", self.select_from)) |
| 156 | |
| 157 | expr = self._apply_expression_modifiers(expr) |
no test coverage detected