/usr/lib/python2.7/dist-packages/sqlalchemy_utils/functions/sort_query.py is in python-sqlalchemy-utils 0.30.12-4.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 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 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 | import sqlalchemy as sa
from sqlalchemy.sql.expression import asc, desc
from .database import has_unique_index
from .orm import get_query_descriptor, get_tables
class QuerySorterException(Exception):
pass
class QuerySorter(object):
def __init__(self, silent=True, separator='-'):
self.separator = separator
self.silent = silent
def assign_order_by(self, entity, attr, func):
expr = get_query_descriptor(self.query, entity, attr)
if expr is not None:
return self.query.order_by(func(expr))
if not self.silent:
raise QuerySorterException(
"Could not sort query with expression '%s'" % attr
)
return self.query
def parse_sort_arg(self, arg):
if arg[0] == self.separator:
func = desc
arg = arg[1:]
else:
func = asc
parts = arg.split(self.separator)
return {
'entity': parts[0] if len(parts) > 1 else None,
'attr': parts[1] if len(parts) > 1 else arg,
'func': func
}
def __call__(self, query, *args):
self.query = query
for sort in args:
if not sort:
continue
self.query = self.assign_order_by(
**self.parse_sort_arg(sort)
)
return self.query
def sort_query(query, *args, **kwargs):
"""
Applies an sql ORDER BY for given query. This function can be easily used
with user-defined sorting.
The examples use the following model definition:
::
import sqlalchemy as sa
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy_utils import sort_query
engine = create_engine(
'sqlite:///'
)
Base = declarative_base()
Session = sessionmaker(bind=engine)
session = Session()
class Category(Base):
__tablename__ = 'category'
id = sa.Column(sa.Integer, primary_key=True)
name = sa.Column(sa.Unicode(255))
class Article(Base):
__tablename__ = 'article'
id = sa.Column(sa.Integer, primary_key=True)
name = sa.Column(sa.Unicode(255))
category_id = sa.Column(sa.Integer, sa.ForeignKey(Category.id))
category = sa.orm.relationship(
Category, primaryjoin=category_id == Category.id
)
1. Applying simple ascending sort
::
query = session.query(Article)
query = sort_query(query, 'name')
2. Appying descending sort
::
query = sort_query(query, '-name')
3. Applying sort to custom calculated label
::
query = session.query(
Category, sa.func.count(Article.id).label('articles')
)
query = sort_query(query, 'articles')
4. Applying sort to joined table column
::
query = session.query(Article).join(Article.category)
query = sort_query(query, 'category-name')
:param query:
query to be modified
:param sort:
string that defines the label or column to sort the query by
:param silent:
Whether or not to raise exceptions if unknown sort column
is passed. By default this is `True` indicating that no errors should
be raised for unknown columns.
"""
return QuerySorter(**kwargs)(query, *args)
def make_order_by_deterministic(query):
"""
Make query order by deterministic (if it isn't already). Order by is
considered deterministic if it contains column that is unique index (
either it is a primary key or has a unique index). Many times it is design
flaw to order by queries in nondeterministic manner.
Consider a User model with three fields: id (primary key), favorite color
and email (unique).::
from sqlalchemy_utils import make_order_by_deterministic
query = session.query(User).order_by(User.favorite_color)
query = make_order_by_deterministic(query)
print query # 'SELECT ... ORDER BY "user".favorite_color, "user".id'
query = session.query(User).order_by(User.email)
query = make_order_by_deterministic(query)
print query # 'SELECT ... ORDER BY "user".email'
query = session.query(User).order_by(User.id)
query = make_order_by_deterministic(query)
print query # 'SELECT ... ORDER BY "user".id'
.. versionadded: 0.27.1
"""
order_by_func = sa.asc
if not query._order_by:
column = None
else:
order_by = query._order_by[0]
if isinstance(order_by, sa.sql.expression.UnaryExpression):
if order_by.modifier == sa.sql.operators.desc_op:
order_by_func = sa.desc
else:
order_by_func = sa.asc
column = order_by.get_children()[0]
else:
column = order_by
# Skip queries that are ordered by an already deterministic column
if isinstance(column, sa.Column):
try:
if has_unique_index(column):
return query
except TypeError:
pass
base_table = get_tables(query._entities[0])[0]
query = query.order_by(
*(order_by_func(c) for c in base_table.c if c.primary_key)
)
return query
|