Hide keyboard shortcuts

Hot-keys on this page

r m x p   toggle line displays

j k   next/prev highlighted chunk

0   (zero) top of page

1   (one) first highlighted chunk

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

# -*- encoding: utf-8 -*- 

 

from sqlalchemy import and_, or_, not_, asc, desc 

 

from filterparams.obj import ( 

    Parameter, 

    BindingOperation, 

    And, 

    Or, 

    Not, 

) 

 

from .util import convert 

 

 

class Evaluation: 

 

    def __init__(self, 

                 query_config): 

        self.query_config = query_config 

 

    @property 

    def query(self): 

        return self.query_config.query 

 

    def evaluate(self, sql_query): 

        sql_query = self._apply_param_order(sql_query) 

        sql_query = self._apply_sort(sql_query) 

        return sql_query 

 

    def _apply_param_order(self, sql_query): 

        if self.query.param_order is not None: 

            item, sql_query = self._process_param_item( 

                self.query.param_order, 

                sql_query, 

            ) 

 

            sql_query = sql_query.filter( 

                item 

            ) 

 

        return sql_query 

 

    def _process_param_item(self, item, sql_query): 

        if isinstance(item, BindingOperation): 

            return self._process_binding_item(item, sql_query) 

        elif isinstance(item, Not): 

            inner_item, sql_query = self._process_param_item( 

                item.inner, sql_query, 

            ) 

            return not_(inner_item), sql_query 

        elif isinstance(item, Parameter): 

            return self._process_parameter(item, sql_query) 

        else: 

            raise ValueError('Unknown parameter %s' % item) 

 

    def _process_binding_item(self, item, sql_query): 

        left_expression, sql_query = self._process_param_item( 

            item.left, 

            sql_query, 

        ) 

        right_expression, sql_query = self._process_param_item( 

            item.right, 

            sql_query, 

        ) 

 

        if isinstance(item, And): 

            result = and_( 

                left_expression, 

                right_expression, 

            ) 

        elif isinstance(item, Or): 

            result = or_( 

                left_expression, 

                right_expression, 

            ) 

        else: 

            raise ValueError("Unkown param entry %s" % item) 

        return result, sql_query 

 

    def _process_parameter(self, item, sql_query): 

        expressions = self.query_config.expressions 

        sql_query, expression = expressions.get_filter_expression( 

            sql_query, 

            item.name 

        ) 

        value = convert( 

            item.value, 

            expression.type, 

            self.query_config.converters 

        ) 

        filter_obj = self.query_config.filter_for(item.filter) 

        return filter_obj(expression, value), sql_query 

 

    def _apply_sort(self, sql_query): 

        expressions = self.query_config.expressions 

 

        orders = [] 

        for sort_item in self.query.orders: 

            sort_func = asc 

            if sort_item.direction == 'desc': 

                sort_func = desc 

            sql_query, expression = expressions.get_filter_expression( 

                sql_query, 

                sort_item.name 

            ) 

            orders.append(sort_func(expression)) 

        return sql_query.order_by(*orders)