})
"""
- try:
- whens = util.dictlike_iteritems(whens)
- except TypeError:
- pass
-
- if value:
- crit_filter = _literal_as_binds
- else:
- crit_filter = _no_literals
-
- whenlist = [ClauseList('WHEN', crit_filter(c), 'THEN', _literal_as_binds(r), operator=None)
- for (c,r) in whens]
- if else_ is not None:
- whenlist.append(ClauseList('ELSE', _literal_as_binds(else_), operator=None))
- if whenlist:
- type = list(whenlist[-1])[-1].type
- else:
- type = None
- cc = _CalculatedClause(None, 'CASE', value, type_=type, operator=None, group_contents=False, *whenlist + ['END'])
- return cc
+
+ return _Case(whens, value=value, else_=else_)
def cast(clause, totype, **kwargs):
"""Return a ``CAST`` function.
"""Return the clause ``expression COLLATE collation``."""
expr = _literal_as_binds(expression)
- return _CalculatedClause(
- expr, expr, _literal_as_text(collation),
- operator=operators.collate, group=False)
+ return _BinaryExpression(
+ expr,
+ _literal_as_text(collation),
+ operators.collate, type_=expr.type)
def exists(*args, **kwargs):
"""Return an ``EXISTS`` clause as applied to a :class:`~sqlalchemy.sql.expression.Select` object.
def collate(self, collation):
"""Produce a COLLATE clause, i.e. ``<column> COLLATE utf8_bin``"""
- return _CalculatedClause(
- None, self, _literal_as_text(collation),
- operator=operators.collate, group=False)
+ return collate(self, collation)
def op(self, operator):
"""produce a generic operator function.
return lambda other: self.__operate(operator, other)
def _bind_param(self, obj):
+ # ONE COmpareMixin
return _BindParamClause(None, obj, type_=self.type, unique=True)
def _check_literal(self, other):
return (self, )
-class _CalculatedClause(ColumnElement):
- """Describe a calculated SQL expression that has a type, like ``CASE``.
-
- Extends ``ColumnElement`` to provide column-level comparison
- operators.
+class _Case(ColumnElement):
+ __visit_name__ = 'case'
- """
-
- __visit_name__ = 'calculatedclause'
+ def __init__(self, whens, value=None, else_=None):
+ try:
+ whens = util.dictlike_iteritems(whens)
+ except TypeError:
+ pass
- def __init__(self, name, *clauses, **kwargs):
- self.name = name
- self.type = sqltypes.to_instance(kwargs.get('type_', None))
- self._bind = kwargs.get('bind', None)
- self.group = kwargs.pop('group', True)
- clauses = ClauseList(
- operator=kwargs.get('operator', None),
- group_contents=kwargs.get('group_contents', True),
- *clauses)
- if self.group:
- self.clause_expr = clauses.self_group()
+ if value:
+ whenlist = [(_literal_as_binds(c).self_group(), _literal_as_binds(r)) for (c, r) in whens]
else:
- self.clause_expr = clauses
-
- @property
- def key(self):
- return self.name or '_calc_'
+ whenlist = [(_no_literals(c).self_group(), _literal_as_binds(r)) for (c, r) in whens]
+
+ if whenlist:
+ type_ = list(whenlist[-1])[-1].type
+ else:
+ type_ = None
+
+ self.value = value
+ self.type = type_
+ self.whens = whenlist
+ if else_ is not None:
+ self.else_ = _literal_as_binds(else_)
+ else:
+ self.else_ = None
def _copy_internals(self, clone=_clone):
- self.clause_expr = clone(self.clause_expr)
-
- @property
- def clauses(self):
- if isinstance(self.clause_expr, _Grouping):
- return self.clause_expr.element
- else:
- return self.clause_expr
+ if self.value:
+ self.value = clone(self.value)
+ self.whens = [(clone(x), clone(y)) for x, y in self.whens]
+ if self.else_:
+ self.else_ = clone(self.else_)
def get_children(self, **kwargs):
- return self.clause_expr,
+ if self.value:
+ yield self.value
+ for x, y in self.whens:
+ yield x
+ yield y
+ if self.else_:
+ yield self.else_
@property
def _from_objects(self):
- return self.clauses._from_objects
+ return itertools.chain(*[x._from_objects for x in self.get_children()])
- def _bind_param(self, obj):
- return _BindParamClause(self.name, obj, type_=self.type, unique=True)
-
- def select(self):
- return select([self])
-
- def scalar(self):
- return select([self]).execute().scalar()
-
- def execute(self):
- return select([self]).execute()
-
- def _compare_type(self, obj):
- return self.type
-
-class Function(_CalculatedClause, FromClause):
- """Describe a SQL function.
-
- Extends ``_CalculatedClause``, turn the *clauselist* into function
- arguments, also adds a `packagenames` argument.
-
- """
+class Function(ColumnElement, FromClause):
+ """Describe a SQL function."""
__visit_name__ = 'function'
def columns(self):
return [self]
+ @util.memoized_property
+ def clauses(self):
+ return self.clause_expr.element
+
+ @property
+ def _from_objects(self):
+ return self.clauses._from_objects
+
+ def get_children(self, **kwargs):
+ return self.clause_expr,
+
def _copy_internals(self, clone=_clone):
- _CalculatedClause._copy_internals(self, clone=clone)
+ self.clause_expr = clone(self.clause_expr)
self._reset_exported()
+ util.reset_memoized(self, 'clauses')
+
+ def _bind_param(self, obj):
+ return _BindParamClause(self.name, obj, type_=self.type, unique=True)
- def get_children(self, **kwargs):
- return _CalculatedClause.get_children(self, **kwargs)
+ def select(self):
+ return select([self])
+
+ def scalar(self):
+ return select([self]).execute().scalar()
+
+ def execute(self):
+ return select([self]).execute()
+
+ def _compare_type(self, obj):
+ return self.type
class _Cast(ColumnElement):
return []
def _bind_param(self, obj):
+ # THREE ColumnCluase
return _BindParamClause(self.name, obj, type_=self.type, unique=True)
def _make_proxy(self, selectable, name=None, attach=True):
select([x.label('foo')]),
'SELECT a AND b AND c AS foo'
)
-
+
self.assert_compile(
and_(table1.c.myid == 12, table1.c.name=='asdf', table2.c.othername == 'foo', "sysdate() = today()"),
"mytable.myid = :myid_1 AND mytable.name = :name_1 "\
self.assert_compile(select([extract("day", func.to_date("03/20/2005", "MM/DD/YYYY"))]), "SELECT extract(day FROM to_date(:to_date_1, :to_date_2)) AS extract_1")
def test_collate(self):
- for expr in (select([table1.c.name.collate('somecol')]),
- select([collate(table1.c.name, 'somecol')])):
+ for expr in (select([table1.c.name.collate('latin1_german2_ci')]),
+ select([collate(table1.c.name, 'latin1_german2_ci')])):
self.assert_compile(
- expr, "SELECT mytable.name COLLATE somecol FROM mytable")
+ expr, "SELECT mytable.name COLLATE latin1_german2_ci AS anon_1 FROM mytable")
+
+ assert table1.c.name.collate('latin1_german2_ci').type is table1.c.name.type
+
+ expr = select([table1.c.name.collate('latin1_german2_ci').label('k1')]).order_by('k1')
+ self.assert_compile(expr,"SELECT mytable.name COLLATE latin1_german2_ci AS k1 FROM mytable ORDER BY k1")
- expr = select([table1.c.name.collate('somecol').like('%x%')])
+ expr = select([collate('foo', 'latin1_german2_ci').label('k1')])
+ self.assert_compile(expr,"SELECT :param_1 COLLATE latin1_german2_ci AS k1")
+
+ expr = select([table1.c.name.collate('latin1_german2_ci').like('%x%')])
self.assert_compile(expr,
- "SELECT mytable.name COLLATE somecol "
+ "SELECT mytable.name COLLATE latin1_german2_ci "
"LIKE :param_1 AS anon_1 FROM mytable")
- expr = select([table1.c.name.like(collate('%x%', 'somecol'))])
+ expr = select([table1.c.name.like(collate('%x%', 'latin1_german2_ci'))])
self.assert_compile(expr,
"SELECT mytable.name "
- "LIKE :param_1 COLLATE somecol AS anon_1 "
+ "LIKE :param_1 COLLATE latin1_german2_ci AS anon_1 "
"FROM mytable")
expr = select([table1.c.name.collate('col1').like(
"LIKE :param_1 COLLATE col2 AS anon_1 "
"FROM mytable")
- expr = select([func.concat('a', 'b').collate('somecol').label('x')])
+ expr = select([func.concat('a', 'b').collate('latin1_german2_ci').label('x')])
self.assert_compile(expr,
"SELECT concat(:param_1, :param_2) "
- "COLLATE somecol AS x")
+ "COLLATE latin1_german2_ci AS x")
+
+
+ expr = select([table1.c.name]).order_by(table1.c.name.collate('latin1_german2_ci'))
+ self.assert_compile(expr, "SELECT mytable.name FROM mytable ORDER BY mytable.name COLLATE latin1_german2_ci")
def test_percent_chars(self):
t = table("table%name",
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
