from dataclasses import dataclass from flask import request, url_for from sqlalchemy import desc from sqlalchemy.orm import Query from sqlalchemy.schema import Column @dataclass class OrderingCriteria: field: Column is_active: bool = False is_desc: bool = True @dataclass class OrderingLink: field: str url: str is_active: bool is_desc: bool def __str__(self): return self.url class QueryOrdering: """ Ordering query by provided ordering string. Desc order direction is providing as '-' before the field name Make sure that all joins are added when ordering by related fields """ def __init__( self, query, allowed_fields=None, order_string="", fields_mapping=None, default_is_desc=False, ): self._query = query self._fields_mapping = fields_mapping or {} self._default_is_desc = default_is_desc self._order_string = order_string or request.args.get( "order", type=str, default="" ) self._criteria = [OrderingCriteria(f) for f in allowed_fields] self._links = {} self._parse_order() self._update_query() self._build_links() @property def links(self): return self._links @property def query(self): return self._query @property def order_string(self): return self._order_string def _parse_order(self): is_active = False for item in self._order_string.split(","): field = item.lstrip("-") field = self._fields_mapping.get(field, field) is_desc = item.startswith("-") for c in self._criteria: if c.field.key == field: c.is_desc = is_desc c.is_active = True is_active = True # activating first default order if nothing passed if not is_active and self._criteria: self._criteria[0].is_active = True self._criteria[0].is_desc = self._default_is_desc def _update_query(self): if not isinstance(self._query, Query): return for c in self._criteria: if not c.is_active: continue if c.is_desc: self._query = self._query.order_by(desc(c.field)) else: self._query = self._query.order_by(c.field) def _build_links(self): reverse_mapping = {v: k for k, v in self._fields_mapping.items()} for c in self._criteria: args = request.args.copy() key = reverse_mapping.get(c.field.key, c.field.key) # Reversing existing if c.is_desc: if c.is_active: args["order"] = "" else: args["order"] = key else: args["order"] = f"-{key}" args = {k: v for k, v in args.items() if v} url = url_for(request.endpoint, **request.view_args, **args) self._links[key] = OrderingLink(key, url, c.is_active, c.is_desc) def __eq__(self, other): try: return str(self.query) == str(other.query) except AttributeError: return False