=======
CHANGES
=======
+0.5.3
+=====
+- orm
+ - Query now implements __clause_element__() which produces
+ its selectable, which means a Query instance can be accepted
+ in many SQL expressions, including col.in_(query),
+ union(query1, query2), select([foo]).select_from(query),
+ etc.
+
+- sql
+ - the __selectable__() interface has been replaced entirely
+ by __clause_element__().
+
0.5.2
======
def update(cls, whereclause=None, values=None, **kwargs):
_ddl_error(cls)
- def __selectable__(cls):
+ def __clause_element__(cls):
return cls._table
def __getattr__(cls, attr):
return x
def class_for_table(selectable, **mapper_kwargs):
- selectable = expression._selectable(selectable)
+ selectable = expression._clause_element_as_expr(selectable)
mapname = 'Mapped' + _selectable_name(selectable)
if isinstance(mapname, unicode):
engine_encoding = selectable.metadata.bind.dialect.encoding
def with_labels(self, item):
# TODO give meaningful aliases
- return self.map(expression._selectable(item).select(use_labels=True).alias('foo'))
+ return self.map(expression._clause_element_as_expr(item).select(use_labels=True).alias('foo'))
def join(self, *args, **kwargs):
j = join(*args, **kwargs)
return self._compile_context(labels=self._with_labels).statement._annotate({'_halt_adapt': True})
- @property
- def _nested_statement(self):
- return self.with_labels().enable_eagerloads(False).statement.correlate(None)
-
def subquery(self):
"""return the full SELECT statement represented by this Query, embedded within an Alias.
Eager JOIN generation within the query is disabled.
"""
-
return self.enable_eagerloads(False).statement.alias()
+
+ def __clause_element__(self):
+ return self.enable_eagerloads(False).statement
@_generative()
def enable_eagerloads(self, value):
those being selected.
"""
- fromclause = self._nested_statement
+ fromclause = self.with_labels().enable_eagerloads(False).statement.correlate(None)
q = self._from_selectable(fromclause)
if entities:
q._set_entities(entities)
"""
return self._from_selectable(
- expression.union(*([self._nested_statement]+ [x._nested_statement for x in q])))
+ expression.union(*([self]+ list(q))))
def union_all(self, *q):
"""Produce a UNION ALL of this Query against one or more queries.
"""
return self._from_selectable(
- expression.union_all(*([self._nested_statement]+ [x._nested_statement for x in q]))
+ expression.union_all(*([self]+ list(q)))
)
def intersect(self, *q):
"""
return self._from_selectable(
- expression.intersect(*([self._nested_statement]+ [x._nested_statement for x in q]))
+ expression.intersect(*([self]+ list(q)))
)
def intersect_all(self, *q):
"""
return self._from_selectable(
- expression.intersect_all(*([self._nested_statement]+ [x._nested_statement for x in q]))
+ expression.intersect_all(*([self]+ list(q)))
)
def except_(self, *q):
"""
return self._from_selectable(
- expression.except_(*([self._nested_statement]+ [x._nested_statement for x in q]))
+ expression.except_(*([self]+ list(q)))
)
def except_all(self, *q):
"""
return self._from_selectable(
- expression.except_all(*([self._nested_statement]+ [x._nested_statement for x in q]))
+ expression.except_all(*([self]+ list(q)))
)
@util.accepts_a_list_as_starargs(list_deprecation='pending')
else:
return element
+def _clause_element_as_expr(element):
+ if hasattr(element, '__clause_element__'):
+ return element.__clause_element__()
+ else:
+ return element
+
def _literal_as_column(element):
if hasattr(element, '__clause_element__'):
return element.__clause_element__()
% (column, getattr(column, 'table', None), fromclause.description))
return c
-def _selectable(element):
- if hasattr(element, '__selectable__'):
- return element.__selectable__()
- elif isinstance(element, Selectable):
- return element
- else:
- raise exc.ArgumentError("Object %r is not a Selectable and does not implement `__selectable__()`" % element)
-
def is_column(col):
"""True if ``col`` is an instance of ``ColumnElement``."""
return isinstance(col, ColumnElement)
return self._in_impl(operators.in_op, operators.notin_op, other)
def _in_impl(self, op, negate_op, seq_or_selectable):
+ seq_or_selectable = _clause_element_as_expr(seq_or_selectable)
+
if isinstance(seq_or_selectable, _ScalarSelect):
return self.__compare( op, seq_or_selectable, negate=negate_op)
__visit_name__ = 'join'
def __init__(self, left, right, onclause=None, isouter=False):
- self.left = _selectable(left)
- self.right = _selectable(right).self_group()
+ self.left = _literal_as_text(left)
+ self.right = _literal_as_text(right).self_group()
if onclause is None:
self.onclause = self._match_primaries(self.left, self.right)
# some DBs do not like ORDER BY in the inner queries of a UNION, etc.
for n, s in enumerate(selects):
+ s = _clause_element_as_expr(s)
+
if not numcols:
numcols = len(s.c)
elif len(s.c) != numcols:
"""return a new select() construct with the given FROM expression applied to its list of
FROM objects."""
- if _is_literal(fromclause):
- fromclause = _TextClause(fromclause)
-
+ fromclause = _literal_as_text(fromclause)
self._froms = self._froms.union([fromclause])
@_generative
self.assert_compile(sess.query(x).filter(x==5).statement,
"SELECT lala(users.id) AS foo FROM users WHERE lala(users.id) = :param_1", dialect=default.DefaultDialect())
-class CompileTest(QueryTest):
+class ExpressionTest(QueryTest):
- def test_deferred(self):
+ def test_deferred_instances(self):
session = create_session()
s = session.query(User).filter(and_(addresses.c.email_address == bindparam('emailad'), Address.user_id==User.id)).statement
l = list(session.query(User).instances(s.execute(emailad = 'jack@bean.com')))
- assert [User(id=7)] == l
+ eq_([User(id=7)], l)
+
+ def test_in(self):
+ session = create_session()
+ s = session.query(User.id).join(User.addresses).group_by(User.id).having(func.count(Address.id) > 2)
+ eq_(
+ session.query(User).filter(User.id.in_(s)).all(),
+ [User(id=8)]
+ )
+
+ def test_union(self):
+ s = create_session()
+
+ q1 = s.query(User).filter(User.name=='ed')
+ q2 = s.query(User).filter(User.name=='fred')
+ eq_(
+ s.query(User).from_statement(union(q1, q2).order_by(User.name)).all(),
+ [User(name='ed'), User(name='fred')]
+ )
+
+ def test_select(self):
+ s = create_session()
+
+ q1 = s.query(User).filter(User.name=='ed')
+ eq_(
+ s.query(User).from_statement(select([q1])).all(),
+ [User(name='ed')]
+ )
+
+ def test_join(self):
+ s = create_session()
+
+ # TODO: do we want aliased() to detect a query and convert to subquery()
+ # automatically ?
+ q1 = s.query(Address).filter(Address.email_address=='jack@bean.com')
+ adalias = aliased(Address, q1.subquery())
+ eq_(
+ s.query(User, adalias).join((adalias, User.id==adalias.user_id)).all(),
+ [(User(id=7,name=u'jack'), Address(email_address=u'jack@bean.com',user_id=7,id=1))]
+ )
+
# more slice tests are available in test/orm/generative.py
class SliceTest(QueryTest):
def test_first(self):
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
