coding utf8 from sqlalchemy import or_ and_ from sqlalchemy sql import

 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
#coding:utf8
from sqlalchemy import or_, and_
from sqlalchemy.sql import operators
from basilisk.tcwfcoll.elements.IChart import Attributes as TChartAttributes
from deliveryrpc.utils.string import normalize
from deliveryrpc.exceptions.filter import (
FilterUnknown,
FilterExpressions
)
OPERATORS_MAP = {
'=': 'eq',
'<': 'lt',
'<=': 'le',
'>': 'gt',
'>=': 'ge',
'<>': 'ne',
'!=': 'ne',
}
IN_PARAMS_MAP = {
'operator': 'op',
'key': 'fieldId',
'value': 'value',
}
class FilterParser(object):
def __init__(self, conditions, user_attributes):
self.conditions = conditions
self.user_attributes = user_attributes
self.binding = None
def get_expr(self, operator, key, value):
operator = getattr(operators, OPERATORS_MAP[operator], None)
if operator is None:
raise FilterUnknown(u'Unknown operator.')
expr = and_(self.user_attributes.Key == key,
operator(self.user_attributes.Value, value))
field = getattr(TChartAttributes, key.lower(), None)
if field is not None:
expr = or_(expr, (operator(field, value)))
return expr
def parse_conditions(self, x):
#binding = unicode(x['binding'], 'utf8')
binding = 'or'
operator = normalize(x[IN_PARAMS_MAP['operator']])
key = normalize(x[IN_PARAMS_MAP['key']])
value = normalize(x[IN_PARAMS_MAP['value']])
expr = self.get_expr(operator, key, value)
binding_func = getattr(operators, binding, None)
if binding_func is None:
binding_func = getattr(operators, '%s_' % binding, None)
if binding_func is None:
raise FilterUnknown(u'Unknown binding.')
if self.binding is None:
self.binding = binding_func
return expr
def get_filters(self):
expr_count = len(self.conditions)
if not expr_count:
raise FilterExpressions(u'Need at least one expression.')
if expr_count == 1:
return [self.parse_conditions(self.conditions[0]),]
expr_list = []
for x in self.conditions:
expr_list.append(self.parse_conditions(x))
return [self.binding(*expr_list), ]