Prodream/ormar
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

try with min and max filters depending on direction

Browse Source
This commit is contained in:
collerek
2021-03-07 17:37:32 +01:00
parent 9c46383130
commit d067c03130
2 changed files with 22 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
ormar/queryset/actions/order_action.py
View File

@ -46,6 +46,13 @@ class OrderAction(QueryAction):
prefix = f"{self.table_prefix}_" if self.table_prefix else ""
return f"{prefix}{self.table}" f".{self.field_alias}"
def get_min_or_max(self):
prefix = f"{self.table_prefix}_" if self.table_prefix else ""
if self.direction == '':
return text(f"min({prefix}{self.table}" f".{self.field_alias})")
else:
return text(f"max({prefix}{self.table}" f".{self.field_alias}) desc")
def get_text_clause(self) -> sqlalchemy.sql.expression.TextClause:
"""
Escapes characters if it's required.

31
ormar/queryset/query.py
View File

@ -171,32 +171,31 @@ class Query:
"""
pk_alias = self.model_cls.get_column_alias(self.model_cls.Meta.pkname)
pk_aliased_name = f"{self.table.name}.{pk_alias}"
qry_text = sqlalchemy.text(f"{pk_aliased_name} as limit_column")
maxes = dict()
for ind, order in enumerate(list(self.sorted_orders.keys())):
qry_text = sqlalchemy.text(f"{pk_aliased_name}")
maxes = OrderedDict()
for order in list(self.sorted_orders.keys()):
if order is not None and order.get_field_name_text() != pk_aliased_name:
aliased_col = order.get_field_name_text()
maxes[aliased_col] = sqlalchemy.text(f"{aliased_col} as col{ind}")
maxes[aliased_col] = order.get_min_or_max()
elif order.get_field_name_text() == pk_aliased_name:
maxes[pk_aliased_name] = order.get_text_clause()
limit_qry = sqlalchemy.sql.select([qry_text] + list(maxes.values()))
limit_qry = sqlalchemy.sql.select([qry_text])
limit_qry = limit_qry.select_from(self.select_from)
limit_qry = self._apply_expression_modifiers(limit_qry)
limit_qry = FilterQuery(filter_clauses=self.filter_clauses).apply(limit_qry)
limit_qry = FilterQuery(filter_clauses=self.exclude_clauses,
exclude=True).apply(
limit_qry
)
limit_qry = OrderQuery(sorted_orders=self.sorted_orders).apply(limit_qry)
limit_qry = limit_qry.alias("inner_limit_query")
outer_text = sqlalchemy.text(f"distinct limit_column")
outer_qry = sqlalchemy.sql.select([outer_text])
outer_qry = outer_qry.select_from(limit_qry)
outer_qry = LimitQuery(limit_count=self.limit_count).apply(outer_qry)
outer_qry = OffsetQuery(query_offset=self.query_offset).apply(outer_qry)
outer_qry = outer_qry.alias("limit_query")
limit_qry = limit_qry.group_by(qry_text)
for order_by in maxes.values():
limit_qry = limit_qry.order_by(order_by)
limit_qry = LimitQuery(limit_count=self.limit_count).apply(limit_qry)
limit_qry = OffsetQuery(query_offset=self.query_offset).apply(limit_qry)
limit_qry = limit_qry.alias("limit_query")
on_clause = sqlalchemy.text(
f"limit_query.limit_column={self.table.name}.{pk_alias}")
return outer_qry, on_clause
f"limit_query.{pk_alias}={self.table.name}.{pk_alias}")
return limit_qry, on_clause
def _apply_expression_modifiers(
self, expr: sqlalchemy.sql.select