--- /dev/null
+.. change::
+ :tags: change, sql
+ :tickets: 4617
+
+ Added new method :meth:`.SelectBase.subquery`, which creates a subquery
+ that is essentially the same thing as what calling
+ :meth:`.FromClause.alias` has always done, e.g. creates a named subquery.
+ This method is intended to roughly mirror the same role as that of
+ :meth:`.Query.subquery`. The :meth:`.SelectBase.alias` method is
+ being kept for the time being as essentially the same function as that
+ of :meth:`.SelectBase.subquery`.
+
+.. change::
+ :tags: change, orm
+ :tickets: 4617
+
+ The ORM will now warn when asked to coerce a :func:`.select` construct into
+ a subquery implicitly. This occurs within places such as the
+ :meth:`.Query.select_entity_from` and :meth:`.Query.select_from` methods
+ as well as within the :func:`.with_polymorphic` function. When a
+ :class:`.SelectBase` (which is what's produced by :func:`.select`) or
+ :class:`.Query` object is passed directly to these functions and others,
+ the ORM is typically coercing them to be a subquery by calling the
+ :meth:`.SelectBase.alias` method automatically (which is now superceded by
+ the :meth:`.SelectBase.subquery method). The historical reason is that
+ most databases other than SQLite don't allow a SELECT of a SELECT without
+ the inner SELECT being a named subuqery in any case; going forward,
+ SQLAlchemy Core is moving towards no longer considering a SELECT statement
+ that isn't inside a subquery to be a "FROM" clause, that is, an object that
+ can be selected from, in the first place, as part of a larger change to
+ unify the interfaces for :func:`.select` and :meth:`.Query`. The change is
+ intended to encourage code to make explicit those places where these
+ subqueries have normally been implicitly created.
self.concrete = concrete
self.single = False
self.inherits = inherits
- self.local_table = local_table
+ if local_table is not None:
+ self.local_table = coercions.expect(
+ roles.StrictFromClauseRole, local_table
+ )
+ else:
+ self.local_table = None
+
self.inherit_condition = inherit_condition
self.inherit_foreign_keys = inherit_foreign_keys
self._init_properties = properties or {}
else:
self.confirm_deleted_rows = confirm_deleted_rows
- if isinstance(self.local_table, expression.SelectBase):
- raise sa_exc.InvalidRequestError(
- "When mapping against a select() construct, map against "
- "an alias() of the construct instead."
- "This because several databases don't allow a "
- "SELECT from a subquery that does not have an alias."
- )
-
self._set_with_polymorphic(with_polymorphic)
self.polymorphic_load = polymorphic_load
else:
self.with_polymorphic = None
- if isinstance(self.local_table, expression.SelectBase):
- raise sa_exc.InvalidRequestError(
- "When mapping against a select() construct, map against "
- "an alias() of the construct instead."
- "This because several databases don't allow a "
- "SELECT from a subquery that does not have an alias."
- )
-
- if self.with_polymorphic and isinstance(
- self.with_polymorphic[1], expression.SelectBase
- ):
+ if self.with_polymorphic and self.with_polymorphic[1] is not None:
self.with_polymorphic = (
self.with_polymorphic[0],
- self.with_polymorphic[1].alias(),
+ coercions.expect(
+ roles.StrictFromClauseRole,
+ self.with_polymorphic[1],
+ allow_select=True,
+ ),
)
if self.configured:
def _with_polymorphic_args(
self, spec=None, selectable=False, innerjoin=False
):
+ if selectable not in (None, False):
+ selectable = coercions.expect(
+ roles.StrictFromClauseRole, selectable, allow_select=True
+ )
+
if self.with_polymorphic:
if not spec:
spec = self.with_polymorphic[0]
"aliased(), or FromClause instance."
)
else:
- if isinstance(from_obj, expression.SelectBase):
- from_obj = from_obj.alias()
+ from_obj = coercions.expect(
+ roles.StrictFromClauseRole, from_obj, allow_select=True
+ )
if set_base_alias:
select_from_alias = from_obj
fa.append(from_obj)
fromclause = (
self.with_labels()
.enable_eagerloads(False)
- .statement.correlate(None)
+ .correlate(None)
+ .subquery()
+ ._anonymous_fromclause()
)
q = self._from_selectable(fromclause)
q._enable_single_crit = False
def _set_op(self, expr_fn, *q):
return self._from_selectable(
- expr_fn(*([self] + list(q)))
+ expr_fn(*([self] + list(q))).subquery()
)._set_enable_single_crit(False)
def union(self, *q):
used at the start of the query to adapt the existing ``User`` entity::
q = session.query(User).\
- select_entity_from(select_stmt).\
+ select_entity_from(select_stmt.subquery()).\
filter(User.name == 'ed')
Above, the generated SQL will show that the ``User`` entity is
:meth:`.TextClause.columns` method::
text_stmt = text("select id, name from user").columns(
- User.id, User.name)
+ User.id, User.name).subquery()
q = session.query(User).select_entity_from(text_stmt)
:meth:`.Query.select_entity_from` itself accepts an :func:`.aliased`
info.is_aliased_class,
)
if self.property._is_self_referential and not is_aliased_class:
- to_selectable = to_selectable.alias()
+ to_selectable = to_selectable._anonymous_fromclause()
single_crit = target_mapper._single_table_criterion
if single_crit is not None:
)
if self.secondary is not None and alias_secondary:
- criterion = ClauseAdapter(self.secondary.alias()).traverse(
- criterion
- )
+ criterion = ClauseAdapter(
+ self.secondary._anonymous_fromclause()
+ ).traverse(criterion)
criterion = visitors.cloned_traverse(
criterion, {}, {"bindparam": visit_bindparam}
aliased = True
if self._is_self_referential and source_selectable is None:
- dest_selectable = dest_selectable.alias()
+ dest_selectable = dest_selectable._anonymous_fromclause()
aliased = True
else:
aliased = True
if aliased:
if secondary is not None:
- secondary = secondary.alias(flat=True)
+ secondary = secondary._anonymous_fromclause(flat=True)
primary_aliasizer = ClauseAdapter(secondary)
secondary_aliasizer = ClauseAdapter(
dest_selectable, equivalents=self.child_equivalents
from .. import inspection
from .. import sql
from .. import util
+from ..sql import coercions
from ..sql import expression
+from ..sql import roles
from ..sql import util as sql_util
for key in table_map:
table = table_map[key]
- # mysql doesn't like selecting from a select;
- # make it an alias of the select
- if isinstance(table, sql.Select):
- table = table.alias()
- table_map[key] = table
+ table = coercions.expect(
+ roles.StrictFromClauseRole, table, allow_select=True
+ )
+ table_map[key] = table
m = {}
for c in table.c:
):
mapper = _class_to_mapper(cls)
if alias is None:
- alias = mapper._with_polymorphic_selectable.alias(
+ alias = mapper._with_polymorphic_selectable._anonymous_fromclause(
name=name, flat=flat
)
raise sa_exc.ArgumentError(
"adapt_on_names only applies to ORM elements"
)
- return element.alias(name, flat=flat)
+ return element._anonymous_fromclause(name=name, flat=flat)
else:
return AliasedClass(
element,
.. seealso:: :meth:`.Join.alias`
- :param selectable: a table or select() statement that will
+ :param selectable: a table or subquery that will
be used in place of the generated FROM clause. This argument is
required if any of the desired classes use concrete table
inheritance, since SQLAlchemy currently cannot generate UNIONs
classes, selectable, innerjoin=innerjoin
)
if aliased or flat:
- selectable = selectable.alias(flat=flat)
+ selectable = selectable._anonymous_fromclause(flat=flat)
return AliasedClass(
base,
selectable,
class _NoTextCoercion(object):
- def _literal_coercion(self, element, argname=None):
+ def _literal_coercion(self, element, argname=None, **kw):
if isinstance(element, util.string_types) and issubclass(
elements.TextClause, self._role_class
):
def _text_coercion(self, element, argname=None):
return _no_text_coercion(element, argname)
- def _literal_coercion(self, element, argname=None):
+ def _literal_coercion(self, element, argname=None, **kw):
if isinstance(element, util.string_types):
if self._coerce_star and element == "*":
return elements.ColumnClause("*", is_literal=True)
class ExpressionElementImpl(
_ColumnCoercions, RoleImpl, roles.ExpressionElementRole
):
- def _literal_coercion(self, element, name=None, type_=None, argname=None):
+ def _literal_coercion(
+ self, element, name=None, type_=None, argname=None, **kw
+ ):
if element is None:
return elements.Null()
else:
ExpressionElementImpl, RoleImpl, roles.BinaryElementRole
):
def _literal_coercion(
- self, element, expr, operator, bindparam_type=None, argname=None
+ self, element, expr, operator, bindparam_type=None, argname=None, **kw
):
try:
return expr._bind_param(operator, element, type_=bindparam_type)
class ConstExprImpl(RoleImpl, roles.ConstExprRole):
- def _literal_coercion(self, element, argname=None):
+ def _literal_coercion(self, element, argname=None, **kw):
if element is None:
return elements.Null()
elif element is False:
else:
self._raise_for_expected(original_element, argname)
- def _literal_coercion(self, element, argname=None):
+ def _literal_coercion(self, element, argname=None, **kw):
"""coerce the given value to :class:`._truncated_label`.
Existing :class:`._truncated_label` and
self._raise_for_expected(original_element, argname)
+class StrictFromClauseImpl(FromClauseImpl, roles.StrictFromClauseRole):
+ def _implicit_coercions(
+ self,
+ original_element,
+ resolved,
+ argname=None,
+ allow_select=False,
+ **kw
+ ):
+ if resolved._is_select_statement and allow_select:
+ util.warn_deprecated(
+ "Implicit coercion of SELECT and textual SELECT constructs "
+ "into FROM clauses is deprecated; please call .subquery() "
+ "on any Core select or ORM Query object in order to produce a "
+ "subquery object."
+ )
+ return resolved.subquery()
+ else:
+ self._raise_for_expected(original_element, argname)
+
+
+class AnonymizedFromClauseImpl(
+ StrictFromClauseImpl, roles.AnonymizedFromClauseRole
+):
+ def _post_coercion(self, element, flat=False, **kw):
+ return element.alias(flat=flat)
+
+
class DMLSelectImpl(_NoTextCoercion, RoleImpl, roles.DMLSelectRole):
def _implicit_coercions(
self, original_element, resolved, argname=None, **kw
raise NotImplementedError()
+class StrictFromClauseRole(FromClauseRole):
+ # does not allow text() or select() objects
+ pass
+
+
+class AnonymizedFromClauseRole(StrictFromClauseRole):
+ # calls .alias() as a post processor
+
+ def _anonymous_fromclause(self, name=None, flat=False):
+ """A synonym for ``.alias()`` that is only present on objects of this
+ role.
+
+ This is an implicit assurance of the target object being part of the
+ role where anonymous aliasing without any warnings is allowed,
+ as opposed to other kinds of SELECT objects that may or may not have
+ an ``.alias()`` method.
+
+ The method is used by the ORM but is currently semi-private to
+ preserve forwards-compatibility.
+
+ """
+ return self.alias(name=name, flat=flat)
+
+
class CoerceTextStatementRole(SQLRole):
_role_name = "Executable SQL, text() construct, or string statement"
return None
-class Join(FromClause):
+class Join(roles.AnonymizedFromClauseRole, FromClause):
"""represent a ``JOIN`` construct between two :class:`.FromClause`
elements.
)
-class Alias(FromClause):
+class Alias(roles.AnonymizedFromClauseRole, FromClause):
"""Represents an table or selectable alias (AS).
Represents an alias, as typically applied to any table or
self.element = state["element"]
-class TableClause(Immutable, FromClause):
+class TableClause(roles.AnonymizedFromClauseRole, Immutable, FromClause):
"""Represents a minimal "table" construct.
This is a lightweight table object that has only a name and a
def _from_objects(self):
return [self]
+ def subquery(self, name=None):
+ """Return a subquery of this :class:`.SelectBase`.
+
+ A subquery is from a SQL perspective a parentheized, named construct
+ that can be placed in the FROM clause of another SELECT statement.
+
+ Given a SELECT statement such as::
+
+ stmt = select([table.c.id, table.c.name])
+
+ The above statement might look like::
+
+ SELECT table.id, table.name FROM table
+
+ The subquery form by itself renders the same way, however when
+ embedded into the FROM clause of another SELECT statement, it becomes
+ a named sub-element::
+
+ subq = stmt.subquery()
+ new_stmt = select([subq])
+
+ The above renders as::
+
+ SELECT anon_1.id, anon_1.name
+ FROM (SELECT table.id, table.name FROM table) AS anon_1
+
+ Historically, :meth:`.SelectBase.subquery` is equivalent to calling
+ the :meth:`.FromClause.alias` method on a FROM object; however,
+ as a :class:`.SelectBase` object is not directly FROM object,
+ the :meth:`.SelectBase.subquery` method provides clearer semantics.
+
+ .. versionadded:: 1.4
+
+ """
+ return self.alias()
+
class GenerativeSelect(SelectBase):
"""Base class for SELECT statements where additional elements can be
"a": ta.select(
tb.c.id == None, # noqa
from_obj=[ta.outerjoin(tb, onclause=atob)],
- ),
+ ).subquery(),
"b": ta.join(tb, onclause=atob)
.outerjoin(tc, onclause=btoc)
.select(tc.c.id == None)
- .reduce_columns(), # noqa
+ .reduce_columns()
+ .subquery(), # noqa
"c": tc.join(tb, onclause=btoc).join(ta, onclause=atob),
},
"type",
"b": ta.join(tb, onclause=atob)
.outerjoin(tc, onclause=btoc)
.select(tc.c.id == None)
- .reduce_columns(), # noqa
+ .reduce_columns()
+ .subquery(), # noqa
"c": tc.join(tb, onclause=btoc).join(ta, onclause=atob),
},
"type",
if jointype == "join1":
poly_union = polymorphic_union(
{
- "person": people.select(people.c.type == "person"),
+ "person": people.select(
+ people.c.type == "person"
+ ).subquery(),
"manager": join(
people,
managers,
elif jointype == "join2":
poly_union = polymorphic_union(
{
- "person": people.select(people.c.type == "person"),
+ "person": people.select(
+ people.c.type == "person"
+ ).subquery(),
"manager": managers.join(
people, people.c.person_id == managers.c.person_id
),
"manager": managers.join(
people, people.c.person_id == managers.c.person_id
),
- "person": people.select(people.c.type == "person"),
+ "person": people.select(
+ people.c.type == "person"
+ ).subquery(),
},
None,
)
managers,
people.c.person_id == managers.c.person_id,
),
- "person": people.select(people.c.type == "person"),
+ "person": people.select(
+ people.c.type == "person"
+ ).subquery(),
},
None,
)
{
"car": cars.outerjoin(offroad_cars)
.select(offroad_cars.c.car_id == None)
- .reduce_columns(), # noqa
+ .reduce_columns()
+ .subquery(),
"offroad": cars.join(offroad_cars),
},
"type",
[table_Employee, table_Engineer.c.machine],
table_Employee.c.atype == "Engineer",
from_obj=[table_Employee.join(table_Engineer)],
- ),
+ ).subquery(),
"Employee": table_Employee.select(
table_Employee.c.atype == "Employee"
- ),
+ ).subquery(),
},
None,
"pu_employee",
[table_Employee, table_Engineer.c.machine],
table_Employee.c.atype == "Engineer",
from_obj=[table_Employee.join(table_Engineer)],
- ),
+ ).subquery(),
},
None,
"pu_engineer",
{
"BaseItem": base_item_table.select(
base_item_table.c.child_name == "BaseItem"
- ),
+ ).subquery(),
"Item": base_item_table.join(item_table),
},
None,
# a 2-col pk in any case but the leading select has a NULL for the
# "t2id" column
d = util.OrderedDict()
- d["t1"] = t1.select(t1.c.type == "t1")
+ d["t1"] = t1.select(t1.c.type == "t1").subquery()
d["t2"] = t1.join(t2)
pjoin = polymorphic_union(d, None, "pjoin")
# a 2-col pk in any case but the leading select has a NULL for the
# "t2id" column
d = util.OrderedDict()
- d["t1"] = t1.select(t1.c.type == "t1")
+ d["t1"] = t1.select(t1.c.type == "t1").subquery()
d["t2"] = t1.join(t2)
pjoin = polymorphic_union(d, None, "pjoin")
),
"p": self.tables.page.select(
self.tables.page.c.type == "p"
- ),
+ ).subquery(),
},
None,
"page_join",
{
"engineer": people.join(engineers),
"manager": people.join(managers),
- "person": people.select(people.c.type == "person"),
+ "person": people.select(people.c.type == "person").subquery(),
},
None,
"pjoin",
{
"engineer": people.join(engineers),
"manager": people.join(managers),
- "person": people.select(people.c.type == "person"),
+ "person": people.select(
+ people.c.type == "person"
+ ).subquery(),
},
None,
"pjoin",
eq_(
sess.query(Manager)
- .select_from(employees.select().limit(10))
+ .select_entity_from(employees.select().limit(10).subquery())
.all(),
[m1, m2],
)
from sqlalchemy import select
from sqlalchemy import String
from sqlalchemy import testing
+from sqlalchemy import text
from sqlalchemy.ext.declarative import comparable_using
from sqlalchemy.ext.declarative import declarative_base
+from sqlalchemy.orm import aliased
from sqlalchemy.orm import AttributeExtension
from sqlalchemy.orm import attributes
from sqlalchemy.orm import collections
from sqlalchemy.orm import comparable_property
from sqlalchemy.orm import composite
from sqlalchemy.orm import configure_mappers
+from sqlalchemy.orm import contains_eager
from sqlalchemy.orm import create_session
from sqlalchemy.orm import defer
from sqlalchemy.orm import deferred
from sqlalchemy.orm import sessionmaker
from sqlalchemy.orm import synonym
from sqlalchemy.orm import undefer
+from sqlalchemy.orm import with_polymorphic
from sqlalchemy.orm.collections import collection
+from sqlalchemy.orm.util import polymorphic_union
from sqlalchemy.testing import assert_raises
from sqlalchemy.testing import assert_raises_message
from sqlalchemy.testing import assertions
from sqlalchemy.testing import eq_
from sqlalchemy.testing import fixtures
from sqlalchemy.testing import is_
+from sqlalchemy.testing import is_true
from sqlalchemy.testing.schema import Column
from sqlalchemy.testing.schema import Table
from sqlalchemy.testing.util import gc_collect
from sqlalchemy.util.compat import pypy
from . import _fixtures
+from .inheritance import _poly_fixtures
from .test_options import PathTest as OptionsPathTest
from .test_transaction import _LocalFixture
self.assert_(len(s.identity_map) == 0)
-class DeprecatedMapperTest(_fixtures.FixtureTest, AssertsCompiledSQL):
+class DeprecatedQueryTest(_fixtures.FixtureTest, AssertsCompiledSQL):
__dialect__ = "default"
+ run_setup_mappers = "once"
+ run_inserts = "once"
+ run_deletes = None
+
+ @classmethod
+ def setup_mappers(cls):
+ cls._setup_stock_mapping()
+
+ @classmethod
+ def _expect_implicit_subquery(cls):
+ return assertions.expect_deprecated(
+ "Implicit coercion of SELECT and textual SELECT constructs into "
+ r"FROM clauses is deprecated; please call \.subquery\(\) on any "
+ "Core select or ORM Query object in order to produce a "
+ "subquery object."
+ )
+
+ def test_via_textasfrom_select_from(self):
+ User = self.classes.User
+ s = create_session()
+
+ with self._expect_implicit_subquery():
+ eq_(
+ s.query(User)
+ .select_from(
+ text("select * from users").columns(
+ id=Integer, name=String
+ )
+ )
+ .order_by(User.id)
+ .all(),
+ [User(id=7), User(id=8), User(id=9), User(id=10)],
+ )
+
def test_query_as_scalar(self):
- users, User = self.tables.users, self.classes.User
+ User = self.classes.User
- mapper(User, users)
s = Session()
with assertions.expect_deprecated(
r"The Query.as_scalar\(\) method is deprecated and will "
):
s.query(User).as_scalar()
+ def test_select_entity_from_crit(self):
+ User, users = self.classes.User, self.tables.users
+
+ sel = users.select()
+ sess = create_session()
+
+ with self._expect_implicit_subquery():
+ eq_(
+ sess.query(User)
+ .select_entity_from(sel)
+ .filter(User.id.in_([7, 8]))
+ .all(),
+ [User(name="jack", id=7), User(name="ed", id=8)],
+ )
+
+ def test_select_entity_from_select(self):
+ User, users = self.classes.User, self.tables.users
+
+ sess = create_session()
+ with self._expect_implicit_subquery():
+ self.assert_compile(
+ sess.query(User.name).select_entity_from(
+ users.select().where(users.c.id > 5)
+ ),
+ "SELECT anon_1.name AS anon_1_name FROM "
+ "(SELECT users.id AS id, users.name AS name FROM users "
+ "WHERE users.id > :id_1) AS anon_1",
+ )
+
+ def test_select_entity_from_q_statement(self):
+ User = self.classes.User
+
+ sess = create_session()
+
+ q = sess.query(User)
+ with self._expect_implicit_subquery():
+ q = sess.query(User).select_entity_from(q.statement)
+ self.assert_compile(
+ q.filter(User.name == "ed"),
+ "SELECT anon_1.id AS anon_1_id, anon_1.name AS anon_1_name "
+ "FROM (SELECT users.id AS id, users.name AS name FROM "
+ "users) AS anon_1 WHERE anon_1.name = :name_1",
+ )
+
+ def test_select_from_q_statement_no_aliasing(self):
+ User = self.classes.User
+ sess = create_session()
+
+ q = sess.query(User)
+ with self._expect_implicit_subquery():
+ q = sess.query(User).select_from(q.statement)
+ self.assert_compile(
+ q.filter(User.name == "ed"),
+ "SELECT users.id AS users_id, users.name AS users_name "
+ "FROM users, (SELECT users.id AS id, users.name AS name FROM "
+ "users) AS anon_1 WHERE users.name = :name_1",
+ )
+
+ def test_from_alias_three(self):
+ User, addresses, users = (
+ self.classes.User,
+ self.tables.addresses,
+ self.tables.users,
+ )
+
+ query = (
+ users.select(users.c.id == 7)
+ .union(users.select(users.c.id > 7))
+ .alias("ulist")
+ .outerjoin(addresses)
+ .select(
+ use_labels=True, order_by=[text("ulist.id"), addresses.c.id]
+ )
+ )
+ sess = create_session()
+
+ # better way. use select_entity_from()
+ def go():
+ with self._expect_implicit_subquery():
+ result = (
+ sess.query(User)
+ .select_entity_from(query)
+ .options(contains_eager("addresses"))
+ .all()
+ )
+ assert self.static.user_address_result == result
+
+ self.assert_sql_count(testing.db, go, 1)
+
+ def test_from_alias_four(self):
+ User, addresses, users = (
+ self.classes.User,
+ self.tables.addresses,
+ self.tables.users,
+ )
+
+ sess = create_session()
+
+ # same thing, but alias addresses, so that the adapter
+ # generated by select_entity_from() is wrapped within
+ # the adapter created by contains_eager()
+ adalias = addresses.alias()
+ query = (
+ users.select(users.c.id == 7)
+ .union(users.select(users.c.id > 7))
+ .alias("ulist")
+ .outerjoin(adalias)
+ .select(use_labels=True, order_by=[text("ulist.id"), adalias.c.id])
+ )
+
+ def go():
+ with self._expect_implicit_subquery():
+ result = (
+ sess.query(User)
+ .select_entity_from(query)
+ .options(contains_eager("addresses", alias=adalias))
+ .all()
+ )
+ assert self.static.user_address_result == result
+
+ self.assert_sql_count(testing.db, go, 1)
+
+ def test_select(self):
+ addresses, users = self.tables.addresses, self.tables.users
+
+ sess = create_session()
+
+ with self._expect_implicit_subquery():
+ self.assert_compile(
+ sess.query(users)
+ .select_entity_from(users.select())
+ .with_labels()
+ .statement,
+ "SELECT users.id AS users_id, users.name AS users_name "
+ "FROM users, "
+ "(SELECT users.id AS id, users.name AS name FROM users) "
+ "AS anon_1",
+ )
+
+ def test_join(self):
+ users, Address, addresses, User = (
+ self.tables.users,
+ self.classes.Address,
+ self.tables.addresses,
+ self.classes.User,
+ )
+
+ # mapper(User, users, properties={"addresses": relationship(Address)})
+ # mapper(Address, addresses)
+
+ sel = users.select(users.c.id.in_([7, 8]))
+ sess = create_session()
+
+ with self._expect_implicit_subquery():
+ result = (
+ sess.query(User)
+ .select_entity_from(sel)
+ .join("addresses")
+ .add_entity(Address)
+ .order_by(User.id)
+ .order_by(Address.id)
+ .all()
+ )
+
+ eq_(
+ result,
+ [
+ (
+ User(name="jack", id=7),
+ Address(user_id=7, email_address="jack@bean.com", id=1),
+ ),
+ (
+ User(name="ed", id=8),
+ Address(user_id=8, email_address="ed@wood.com", id=2),
+ ),
+ (
+ User(name="ed", id=8),
+ Address(user_id=8, email_address="ed@bettyboop.com", id=3),
+ ),
+ (
+ User(name="ed", id=8),
+ Address(user_id=8, email_address="ed@lala.com", id=4),
+ ),
+ ],
+ )
+
+ adalias = aliased(Address)
+ with self._expect_implicit_subquery():
+ result = (
+ sess.query(User)
+ .select_entity_from(sel)
+ .join(adalias, "addresses")
+ .add_entity(adalias)
+ .order_by(User.id)
+ .order_by(adalias.id)
+ .all()
+ )
+ eq_(
+ result,
+ [
+ (
+ User(name="jack", id=7),
+ Address(user_id=7, email_address="jack@bean.com", id=1),
+ ),
+ (
+ User(name="ed", id=8),
+ Address(user_id=8, email_address="ed@wood.com", id=2),
+ ),
+ (
+ User(name="ed", id=8),
+ Address(user_id=8, email_address="ed@bettyboop.com", id=3),
+ ),
+ (
+ User(name="ed", id=8),
+ Address(user_id=8, email_address="ed@lala.com", id=4),
+ ),
+ ],
+ )
+
+ def test_more_joins(self):
+ (
+ users,
+ Keyword,
+ orders,
+ items,
+ order_items,
+ Order,
+ Item,
+ User,
+ keywords,
+ item_keywords,
+ ) = (
+ self.tables.users,
+ self.classes.Keyword,
+ self.tables.orders,
+ self.tables.items,
+ self.tables.order_items,
+ self.classes.Order,
+ self.classes.Item,
+ self.classes.User,
+ self.tables.keywords,
+ self.tables.item_keywords,
+ )
+
+ sess = create_session()
+ sel = users.select(users.c.id.in_([7, 8]))
+
+ with self._expect_implicit_subquery():
+ eq_(
+ sess.query(User)
+ .select_entity_from(sel)
+ .join("orders", "items", "keywords")
+ .filter(Keyword.name.in_(["red", "big", "round"]))
+ .all(),
+ [User(name="jack", id=7)],
+ )
+
+ with self._expect_implicit_subquery():
+ eq_(
+ sess.query(User)
+ .select_entity_from(sel)
+ .join("orders", "items", "keywords", aliased=True)
+ .filter(Keyword.name.in_(["red", "big", "round"]))
+ .all(),
+ [User(name="jack", id=7)],
+ )
+
+ def test_join_no_order_by(self):
+ User, users = self.classes.User, self.tables.users
+
+ sel = users.select(users.c.id.in_([7, 8]))
+ sess = create_session()
+
+ with self._expect_implicit_subquery():
+ eq_(
+ sess.query(User).select_entity_from(sel).all(),
+ [User(name="jack", id=7), User(name="ed", id=8)],
+ )
+
+ def test_replace_with_eager(self):
+ users, Address, addresses, User = (
+ self.tables.users,
+ self.classes.Address,
+ self.tables.addresses,
+ self.classes.User,
+ )
+
+ sel = users.select(users.c.id.in_([7, 8]))
+ sess = create_session()
+
+ def go():
+ with self._expect_implicit_subquery():
+ eq_(
+ sess.query(User)
+ .options(joinedload("addresses"))
+ .select_entity_from(sel)
+ .order_by(User.id)
+ .all(),
+ [
+ User(id=7, addresses=[Address(id=1)]),
+ User(
+ id=8,
+ addresses=[
+ Address(id=2),
+ Address(id=3),
+ Address(id=4),
+ ],
+ ),
+ ],
+ )
+
+ self.assert_sql_count(testing.db, go, 1)
+ sess.expunge_all()
+
+ def go():
+ with self._expect_implicit_subquery():
+ eq_(
+ sess.query(User)
+ .options(joinedload("addresses"))
+ .select_entity_from(sel)
+ .filter(User.id == 8)
+ .order_by(User.id)
+ .all(),
+ [
+ User(
+ id=8,
+ addresses=[
+ Address(id=2),
+ Address(id=3),
+ Address(id=4),
+ ],
+ )
+ ],
+ )
+
+ self.assert_sql_count(testing.db, go, 1)
+ sess.expunge_all()
+
+ def go():
+ with self._expect_implicit_subquery():
+ eq_(
+ sess.query(User)
+ .options(joinedload("addresses"))
+ .select_entity_from(sel)
+ .order_by(User.id)[1],
+ User(
+ id=8,
+ addresses=[
+ Address(id=2),
+ Address(id=3),
+ Address(id=4),
+ ],
+ ),
+ )
+
+ self.assert_sql_count(testing.db, go, 1)
+
+
+class DeprecatedInhTest(_poly_fixtures._Polymorphic):
+ def test_with_polymorphic(self):
+ Person = _poly_fixtures.Person
+ Engineer = _poly_fixtures.Engineer
+
+ with DeprecatedQueryTest._expect_implicit_subquery():
+ p_poly = with_polymorphic(Person, [Engineer], select([Person]))
+
+ is_true(
+ sa.inspect(p_poly).selectable.compare(select([Person]).subquery())
+ )
+
+
+class DeprecatedMapperTest(_fixtures.FixtureTest, AssertsCompiledSQL):
+ __dialect__ = "default"
+
+ def test_polymorphic_union_w_select(self):
+ users, addresses = self.tables.users, self.tables.addresses
+
+ with DeprecatedQueryTest._expect_implicit_subquery():
+ dep = polymorphic_union(
+ {"u": users.select(), "a": addresses.select()},
+ "type",
+ "bcjoin",
+ )
+
+ subq_version = polymorphic_union(
+ {
+ "u": users.select().subquery(),
+ "a": addresses.select().subquery(),
+ },
+ "type",
+ "bcjoin",
+ )
+ is_true(dep.compare(subq_version))
+
def test_cancel_order_by(self):
users, User = self.tables.users, self.classes.User
sel = users.select(users.c.id.in_([7, 8]))
sess = create_session()
- eq_(
- sess.query(User).select_entity_from(sel).all(),
- [User(name="jack", id=7), User(name="ed", id=8)],
- )
+ with DeprecatedQueryTest._expect_implicit_subquery():
+ eq_(
+ sess.query(User).select_entity_from(sel).all(),
+ [User(name="jack", id=7), User(name="ed", id=8)],
+ )
def test_defer_addtl_attrs(self):
users, User, Address, addresses = (
self.assert_compile(
sess.query(users)
- .select_entity_from(users.select())
+ .select_entity_from(users.select().subquery())
.with_labels()
.statement,
"SELECT users.id AS users_id, users.name AS users_name "
sess = create_session()
q = sess.query(User)
- q = sess.query(User).select_entity_from(q.statement)
+ q = sess.query(User).select_entity_from(q.statement.subquery())
self.assert_compile(
q.filter(User.name == "ed"),
"SELECT anon_1.id AS anon_1_id, anon_1.name AS anon_1_name "
sess = create_session()
q = sess.query(User)
- q = sess.query(User).select_from(q.statement)
+ q = sess.query(User).select_from(q.statement.subquery())
self.assert_compile(
q.filter(User.name == "ed"),
"SELECT users.id AS users_id, users.name AS users_name "
def go():
result = (
sess.query(User)
- .select_entity_from(query)
+ .select_entity_from(query.subquery())
.options(contains_eager("addresses"))
.all()
)
def go():
result = (
sess.query(User)
- .select_entity_from(query)
+ .select_entity_from(query.subquery())
.options(contains_eager("addresses", alias=adalias))
.all()
)
# here for comparison
self.assert_compile(
sess.query(User.name).select_entity_from(
- users.select().where(users.c.id > 5)
+ users.select().where(users.c.id > 5).subquery()
),
"SELECT anon_1.name AS anon_1_name FROM (SELECT users.id AS id, "
"users.name AS name FROM users WHERE users.id > :id_1) AS anon_1",
sess = create_session()
eq_(
- sess.query(User).select_entity_from(sel).all(),
+ sess.query(User).select_entity_from(sel.subquery()).all(),
[User(name="jack", id=7), User(name="ed", id=8)],
)
eq_(
sess.query(User)
- .select_entity_from(sel)
+ .select_entity_from(sel.subquery())
.join("addresses")
.add_entity(Address)
.order_by(User.id)
adalias = aliased(Address)
eq_(
sess.query(User)
- .select_entity_from(sel)
+ .select_entity_from(sel.subquery())
.join(adalias, "addresses")
.add_entity(adalias)
.order_by(User.id)
eq_(
sess.query(User)
- .select_entity_from(sel)
+ .select_entity_from(sel.subquery())
.join("orders", "items", "keywords")
.filter(Keyword.name.in_(["red", "big", "round"]))
.all(),
eq_(
sess.query(User)
- .select_entity_from(sel)
+ .select_entity_from(sel.subquery())
.join("orders", "items", "keywords", aliased=True)
.filter(Keyword.name.in_(["red", "big", "round"]))
.all(),
def go():
eq_(
sess.query(User)
- .select_entity_from(sel)
+ .select_entity_from(sel.subquery())
.options(
joinedload("orders")
.joinedload("items")
sel2 = orders.select(orders.c.id.in_([1, 2, 3]))
eq_(
sess.query(Order)
- .select_entity_from(sel2)
+ .select_entity_from(sel2.subquery())
.join("items", "keywords")
.filter(Keyword.name == "red")
.order_by(Order.id)
)
eq_(
sess.query(Order)
- .select_entity_from(sel2)
+ .select_entity_from(sel2.subquery())
.join("items", "keywords", aliased=True)
.filter(Keyword.name == "red")
.order_by(Order.id)
eq_(
sess.query(User)
.options(joinedload("addresses"))
- .select_entity_from(sel)
+ .select_entity_from(sel.subquery())
.order_by(User.id)
.all(),
[
eq_(
sess.query(User)
.options(joinedload("addresses"))
- .select_entity_from(sel)
+ .select_entity_from(sel.subquery())
.filter(User.id == 8)
.order_by(User.id)
.all(),
eq_(
sess.query(User)
.options(joinedload("addresses"))
- .select_entity_from(sel)
+ .select_entity_from(sel.subquery())
.order_by(User.id)[1],
User(
id=8,
eq_(
s.query(User)
.select_from(
- text("select * from users").columns(id=Integer, name=String)
+ text("select * from users")
+ .columns(id=Integer, name=String)
+ .subquery()
)
.order_by(User.id)
.all(),
Subset, common = self.classes.Subset, self.tables.common
subset_select = select([common.c.id, common.c.data])
- assert_raises(
- sa.exc.InvalidRequestError, mapper, Subset, subset_select
- )
+ assert_raises(sa.exc.ArgumentError, mapper, Subset, subset_select)
def test_basic(self):
Subset, common = self.classes.Subset, self.tables.common
is_true(stmt.compare(select([table1.c.myid]).scalar_subquery()))
+ def test_fromclause_subquery(self):
+ stmt = select([table1.c.myid])
+ with testing.expect_deprecated(
+ "Implicit coercion of SELECT and textual SELECT constructs "
+ "into FROM clauses is deprecated"
+ ):
+ coerced = coercions.expect(
+ roles.StrictFromClauseRole, stmt, allow_select=True
+ )
+
+ is_true(coerced.compare(stmt.subquery()))
+
class TextTest(fixtures.TestBase, AssertsCompiledSQL):
__dialect__ = "default"
d1 = DDL("hi")
is_(expect(roles.CoerceTextStatementRole, d1), d1)
+ def test_strict_from_clause_role(self):
+ stmt = select([t]).subquery()
+ is_true(
+ expect(roles.StrictFromClauseRole, stmt).compare(
+ select([t]).subquery()
+ )
+ )
+
+ def test_strict_from_clause_role_disallow_select(self):
+ stmt = select([t])
+ assert_raises_message(
+ exc.ArgumentError,
+ r"FROM expression, such as a Table or alias\(\) "
+ "object expected, got .*Select",
+ expect,
+ roles.StrictFromClauseRole,
+ stmt,
+ )
+
+ def test_anonymized_from_clause_role(self):
+ is_true(expect(roles.AnonymizedFromClauseRole, t).compare(t.alias()))
+
+ # note the compare for subquery().alias(), even if it is two
+ # plain Alias objects (which it won't be once we introduce the
+ # Subquery class), still compares based on alias() being present
+ # twice, that is, alias().alias() builds an alias of an alias, rather
+ # than just replacing the outer alias.
+ is_true(
+ expect(
+ roles.AnonymizedFromClauseRole, select([t]).subquery()
+ ).compare(select([t]).subquery().alias())
+ )
+
def test_statement_coercion_sequence(self):
s1 = Sequence("hi")
is_(expect(roles.CoerceTextStatementRole, s1), s1)
{
"BaseItem": base_item_table.select(
base_item_table.c.child_name == "BaseItem"
- ),
+ ).subquery(),
"Item": base_item_table.join(item_table),
},
None,
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
