145 lines
4.1 KiB
Python
145 lines
4.1 KiB
Python
import six
|
|
import sqlalchemy as sa
|
|
from sqlalchemy.sql.annotation import AnnotatedColumn
|
|
from sqlalchemy.sql.expression import (
|
|
BooleanClauseList,
|
|
BinaryExpression,
|
|
UnaryExpression,
|
|
BindParameter,
|
|
Cast,
|
|
)
|
|
from sqlalchemy.sql.elements import (
|
|
False_,
|
|
True_,
|
|
Grouping,
|
|
ClauseList,
|
|
Label,
|
|
Case,
|
|
Tuple,
|
|
Null
|
|
)
|
|
|
|
|
|
class ExpressionParser(object):
|
|
def expression(self, expr):
|
|
if expr is None:
|
|
return
|
|
if isinstance(expr, BinaryExpression):
|
|
return self.binary_expression(expr)
|
|
elif isinstance(expr, BooleanClauseList):
|
|
return self.boolean_expression(expr)
|
|
elif isinstance(expr, UnaryExpression):
|
|
return self.unary_expression(expr)
|
|
elif isinstance(expr, sa.Column):
|
|
return self.column(expr)
|
|
elif isinstance(expr, AnnotatedColumn):
|
|
return self.column(expr)
|
|
elif isinstance(expr, BindParameter):
|
|
return self.bind_parameter(expr)
|
|
elif isinstance(expr, False_):
|
|
return self.false(expr)
|
|
elif isinstance(expr, True_):
|
|
return self.true(expr)
|
|
elif isinstance(expr, Grouping):
|
|
return self.grouping(expr)
|
|
elif isinstance(expr, ClauseList):
|
|
return self.clause_list(expr)
|
|
elif isinstance(expr, Label):
|
|
return self.label(expr)
|
|
elif isinstance(expr, Cast):
|
|
return self.cast(expr)
|
|
elif isinstance(expr, Case):
|
|
return self.case(expr)
|
|
elif isinstance(expr, Tuple):
|
|
return self.tuple(expr)
|
|
elif isinstance(expr, Null):
|
|
return self.null(expr)
|
|
raise Exception(
|
|
'Unknown expression type %s' % expr.__class__.__name__
|
|
)
|
|
|
|
def null(self, expr):
|
|
return expr
|
|
|
|
def tuple(self, expr):
|
|
return expr.__class__(
|
|
*map(self.expression, expr.clauses),
|
|
type_=expr.type
|
|
)
|
|
|
|
def clause_list(self, expr):
|
|
return expr.__class__(
|
|
*map(self.expression, expr.clauses),
|
|
group=expr.group,
|
|
group_contents=expr.group_contents,
|
|
operator=expr.operator
|
|
)
|
|
|
|
def label(self, expr):
|
|
return expr.__class__(
|
|
name=expr.name,
|
|
element=self.expression(expr._element),
|
|
type_=expr.type
|
|
)
|
|
|
|
def cast(self, expr):
|
|
return expr.__class__(
|
|
expression=self.expression(expr.clause),
|
|
type_=expr.type
|
|
)
|
|
|
|
def case(self, expr):
|
|
return expr.__class__(
|
|
whens=[
|
|
tuple(self.expression(x) for x in when) for when in expr.whens
|
|
],
|
|
value=self.expression(expr.value),
|
|
else_=self.expression(expr.else_)
|
|
)
|
|
|
|
def grouping(self, expr):
|
|
return expr.__class__(self.expression(expr.element))
|
|
|
|
def true(self, expr):
|
|
return expr
|
|
|
|
def false(self, expr):
|
|
return expr
|
|
|
|
def process_table(self, table):
|
|
return table
|
|
|
|
def column(self, column):
|
|
table = self.process_table(column.table)
|
|
return table.c[column.name]
|
|
|
|
def unary_expression(self, expr):
|
|
return expr.operator(self.expression(expr.element))
|
|
|
|
def bind_parameter(self, expr):
|
|
# somehow bind parameters passed as unicode are converted to
|
|
# ascii strings along the way, force convert them back to avoid
|
|
# sqlalchemy unicode warnings
|
|
if isinstance(expr.type, sa.Unicode):
|
|
expr.value = six.text_type(expr.value)
|
|
return expr
|
|
|
|
def binary_expression(self, expr):
|
|
return expr.__class__(
|
|
left=self.expression(expr.left),
|
|
right=self.expression(expr.right),
|
|
operator=expr.operator,
|
|
type_=expr.type,
|
|
negate=expr.negate,
|
|
modifiers=expr.modifiers.copy()
|
|
)
|
|
|
|
def boolean_expression(self, expr):
|
|
return expr.operator(*[
|
|
self.expression(child_expr)
|
|
for child_expr in expr.get_children()
|
|
])
|
|
|
|
def __call__(self, expr):
|
|
return self.expression(expr)
|