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

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 order_by_clauses = query._order_by_clauses 

44 if not order_by_clauses: 

45 column = None 

46 else: 

47 order_by = order_by_clauses[0] 

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

49 order_by = order_by.element 

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

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

52 order_by_func = sa.desc 

53 else: 

54 order_by_func = sa.asc 

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

56 else: 

57 column = order_by 

58 

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

60 if isinstance(column, sa.Column): 

61 try: 

62 if has_unique_index(column): 

63 return query 

64 except TypeError: 

65 pass 

66 

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

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

69 return query