#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), ]