Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/sqlalchemy_utils/functions/sort_query.py: 14%

1import sqlalchemy as sa 

2 

3from .database import has_unique_index 

4from .orm import _get_query_compile_state, get_tables 

5 

6 

7def make_order_by_deterministic(query): 

8 """ 

9 Make query order by deterministic (if it isn't already). Order by is 

10 considered deterministic if it contains column that is unique index ( 

11 either it is a primary key or has a unique index). Many times it is design 

12 flaw to order by queries in nondeterministic manner. 

13 

14 Consider a User model with three fields: id (primary key), favorite color 

15 and email (unique).:: 

16 

17 

18 from sqlalchemy_utils import make_order_by_deterministic 

19 

20 

21 query = session.query(User).order_by(User.favorite_color) 

22 

23 query = make_order_by_deterministic(query) 

24 print query # 'SELECT ... ORDER BY "user".favorite_color, "user".id' 

25 

26 

27 query = session.query(User).order_by(User.email) 

28 

29 query = make_order_by_deterministic(query) 

30 print query # 'SELECT ... ORDER BY "user".email' 

31 

32 

33 query = session.query(User).order_by(User.id) 

34 

35 query = make_order_by_deterministic(query) 

36 print query # 'SELECT ... ORDER BY "user".id' 

37 

38 

39 .. versionadded: 0.27.1 

40 """ 

41 order_by_func = sa.asc 

42 

43 try: 

44 order_by_clauses = query._order_by_clauses 

45 except AttributeError: # SQLAlchemy <1.4 

46 order_by_clauses = query._order_by 

47 if not order_by_clauses: 

48 column = None 

49 else: 

50 order_by = order_by_clauses[0] 

51 if isinstance(order_by, sa.sql.elements._label_reference): 

52 order_by = order_by.element 

53 if isinstance(order_by, sa.sql.expression.UnaryExpression): 

54 if order_by.modifier == sa.sql.operators.desc_op: 

55 order_by_func = sa.desc 

56 else: 

57 order_by_func = sa.asc 

58 column = list(order_by.get_children())[0] 

59 else: 

60 column = order_by 

61 

62 # Skip queries that are ordered by an already deterministic column 

63 if isinstance(column, sa.Column): 

64 try: 

65 if has_unique_index(column): 

66 return query 

67 except TypeError: 

68 pass 

69 

70 base_table = get_tables(_get_query_compile_state(query)._entities[0])[0] 

71 query = query.order_by(*(order_by_func(c) for c in base_table.c if c.primary_key)) 

72 return query