mixins.py 1.7 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647
  1. from django.core.exceptions import FullResultSet
  2. from django.db.models.expressions import OrderByList
  3. class OrderableAggMixin:
  4. def __init__(self, *expressions, ordering=(), **extra):
  5. if not ordering:
  6. self.order_by = None
  7. elif isinstance(ordering, (list, tuple)):
  8. self.order_by = OrderByList(*ordering)
  9. else:
  10. self.order_by = OrderByList(ordering)
  11. super().__init__(*expressions, **extra)
  12. def resolve_expression(self, *args, **kwargs):
  13. if self.order_by is not None:
  14. self.order_by = self.order_by.resolve_expression(*args, **kwargs)
  15. return super().resolve_expression(*args, **kwargs)
  16. def get_source_expressions(self):
  17. return super().get_source_expressions() + [self.order_by]
  18. def set_source_expressions(self, exprs):
  19. *exprs, self.order_by = exprs
  20. return super().set_source_expressions(exprs)
  21. def as_sql(self, compiler, connection):
  22. *source_exprs, filtering_expr, ordering_expr = self.get_source_expressions()
  23. order_by_sql = ""
  24. order_by_params = []
  25. if ordering_expr is not None:
  26. order_by_sql, order_by_params = compiler.compile(ordering_expr)
  27. filter_params = []
  28. if filtering_expr is not None:
  29. try:
  30. _, filter_params = compiler.compile(filtering_expr)
  31. except FullResultSet:
  32. pass
  33. source_params = []
  34. for source_expr in source_exprs:
  35. source_params += compiler.compile(source_expr)[1]
  36. sql, _ = super().as_sql(compiler, connection, ordering=order_by_sql)
  37. return sql, (*source_params, *order_by_params, *filter_params)