adapters = []
- # vvvvvvvvvvvvvvv legacy vvvvvvvvvvvvvvvvvv
if self._from_obj_alias:
+ # used for legacy going forward for query set_ops, e.g.
+ # union(), union_all(), etc.
+ # 1.4 and previously, also used for from_self(),
+ # select_entity_from()
+ #
# for the "from obj" alias, apply extra rule to the
# 'ORM only' check, if this query were generated from a
# subquery of itself, i.e. _from_selectable(), apply adaption
)
)
+ # vvvvvvvvvvvvvvv legacy vvvvvvvvvvvvvvvvvv
+ # this can totally go away when we remove join(..., aliased=True)
if self._aliased_generations:
adapters.append((False, self._adapt_aliased_generation))
# ^^^^^^^^^^^^^ legacy ^^^^^^^^^^^^^^^^^^^^^
- # this is the only adapter we would need going forward...
+ # this was *hopefully* the only adapter we were going to need
+ # going forward...however, we unfortunately need _from_obj_alias
+ # for query.union(), which we can't drop
if self._polymorphic_adapters:
adapters.append((False, self._adapt_polymorphic_element))
r"The Query\.get\(\) method",
r"The Query\.with_parent\(\) method",
r"The Query\.with_parent\(\) method",
- r"The Query\.select_entity_from\(\) method",
r"The ``aliased`` and ``from_joinpoint`` keyword arguments",
r"The Query.with_polymorphic\(\) method is considered "
"legacy as of the 1.x series",
s = fixture_session()
- q = s.query(ASingleSubA).select_entity_from(select(ASingle).subquery())
+ q = s.query(ASingleSubA).from_statement(select(ASingle))
assert_raises_message(
sa_exc.InvalidRequestError,
s = fixture_session()
- q = s.query(AJoinedSubA).select_entity_from(select(AJoined).subquery())
+ q = s.query(AJoinedSubA).from_statement(select(AJoined))
assert_raises_message(
sa_exc.InvalidRequestError,
sess.add_all([m1, m2, e1, e2])
sess.flush()
- # note test_basic -> UnexpectedPolymorphicIdentityTest as well
+ ma = aliased(
+ Manager,
+ employees.select()
+ .where(employees.c.type == "manager")
+ .order_by(employees.c.employee_id)
+ .limit(10)
+ .subquery(),
+ )
eq_(
- sess.query(Manager)
- .select_entity_from(
- employees.select()
- .where(employees.c.type == "manager")
- .order_by(employees.c.employee_id)
- .limit(10)
- .subquery()
- )
- .all(),
+ sess.query(ma).all(),
[m1, m2],
)
sess.add_all([m1, m2, e1, e2])
sess.flush()
- stmt = (
- select(reports, employees)
- .select_from(
- reports.outerjoin(
- employees,
- and_(
- employees.c.employee_id == reports.c.employee_id,
- employees.c.type == "manager",
- ),
- )
+ stmt = select(reports, employees).select_from(
+ reports.outerjoin(
+ employees,
+ and_(
+ employees.c.employee_id == reports.c.employee_id,
+ employees.c.type == "manager",
+ ),
)
- .set_label_style(LABEL_STYLE_TABLENAME_PLUS_COL)
- .subquery()
)
+
+ subq = stmt.subquery()
+
+ ra = aliased(Report, subq)
+
+ # this test previously used select_entity_from(). the standard
+ # conversion to use aliased() neds to be adjusted to be against
+ # Employee, not Manger, otherwise the ORM will add the manager single
+ # inh criteria to the outside which will break the outer join
+ ma = aliased(Employee, subq)
+
+ eq_(
+ sess.query(ra, ma).order_by(ra.name).all(),
+ [(r1, m1), (r2, m2), (r3, None), (r4, None)],
+ )
+
+ # however if someone really wants to run that SELECT statement and
+ # get back these two entities, they can use from_statement() more
+ # directly. in 1.4 we don't even need tablename label style for the
+ # select(), automatic disambiguation works great
eq_(
sess.query(Report, Manager)
- .select_entity_from(stmt)
- .order_by(Report.name)
+ .from_statement(stmt.order_by(reports.c.name))
.all(),
[(r1, m1), (r2, m2), (r3, None), (r4, None)],
)
from sqlalchemy import func
from sqlalchemy import inspect
from sqlalchemy import Integer
+from sqlalchemy import join
+from sqlalchemy import lateral
from sqlalchemy import literal_column
from sqlalchemy import MetaData
from sqlalchemy import or_
r"in the ORM with_parent\(\) function"
)
+sef_dep = (
+ r"The Query.select_entity_from\(\) method is considered "
+ "legacy as of the 1.x"
+)
+
def _aliased_join_warning(arg=None):
return testing.expect_warnings(
self.assert_sql_count(testing.db, go, 1)
+class TextTest(QueryTest):
+ def test_via_textasfrom_select_from(self):
+ User = self.classes.User
+ s = fixture_session()
+
+ with assertions.expect_deprecated_20(sef_dep):
+ eq_(
+ s.query(User)
+ .select_entity_from(
+ text("select * from users")
+ .columns(User.id, User.name)
+ .subquery()
+ )
+ .order_by(User.id)
+ .all(),
+ [User(id=7), User(id=8), User(id=9), User(id=10)],
+ )
+
+
class TestDeprecation20(fixtures.TestBase):
def test_relation(self):
with testing.expect_deprecated_20(".*relationship"):
with assertions.expect_deprecated_20(opt_strings_dep):
self.assert_sql_count(testing.db, go, 2)
+
+
+class ColumnAccessTest(QueryTest, AssertsCompiledSQL):
+ """test access of columns after _from_selectable has been applied"""
+
+ __dialect__ = "default"
+
+ def test_select_entity_from(self):
+ User = self.classes.User
+ sess = fixture_session()
+
+ q = sess.query(User)
+ with assertions.expect_deprecated_20(sef_dep):
+ 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 "
+ "FROM (SELECT users.id AS id, users.name AS name FROM "
+ "users) AS anon_1 WHERE anon_1.name = :name_1",
+ )
+
+ def test_select_entity_from_no_entities(self):
+ User = self.classes.User
+ sess = fixture_session()
+
+ with assertions.expect_deprecated_20(sef_dep):
+ assert_raises_message(
+ sa.exc.ArgumentError,
+ r"A selectable \(FromClause\) instance is "
+ "expected when the base alias is being set",
+ sess.query(User).select_entity_from(User)._compile_context,
+ )
+
+
+class SelectFromTest(QueryTest, AssertsCompiledSQL):
+ run_setup_mappers = None
+ __dialect__ = "default"
+
+ def test_aliased_class_vs_nonaliased(self):
+ User, users = self.classes.User, self.tables.users
+ self.mapper_registry.map_imperatively(User, users)
+
+ sess = fixture_session()
+ with assertions.expect_deprecated_20(sef_dep):
+ self.assert_compile(
+ sess.query(User.name).select_entity_from(
+ 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",
+ )
+
+ def test_join_no_order_by(self):
+ User, users = self.classes.User, self.tables.users
+
+ self.mapper_registry.map_imperatively(User, users)
+
+ sel = users.select().where(users.c.id.in_([7, 8]))
+ sess = fixture_session()
+
+ with assertions.expect_deprecated_20(sef_dep):
+ eq_(
+ sess.query(User).select_entity_from(sel.subquery()).all(),
+ [User(name="jack", id=7), User(name="ed", id=8)],
+ )
+
+ def test_join(self):
+ users, Address, addresses, User = (
+ self.tables.users,
+ self.classes.Address,
+ self.tables.addresses,
+ self.classes.User,
+ )
+
+ self.mapper_registry.map_imperatively(
+ User, users, properties={"addresses": relationship(Address)}
+ )
+ self.mapper_registry.map_imperatively(Address, addresses)
+
+ sel = users.select().where(users.c.id.in_([7, 8]))
+ sess = fixture_session()
+
+ with assertions.expect_deprecated_20(sef_dep):
+ eq_(
+ sess.query(User)
+ .select_entity_from(sel.subquery())
+ .join("addresses")
+ .add_entity(Address)
+ .order_by(User.id)
+ .order_by(Address.id)
+ .all(),
+ [
+ (
+ 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 assertions.expect_deprecated_20(sef_dep):
+ eq_(
+ sess.query(User)
+ .select_entity_from(sel.subquery())
+ .join(adalias, "addresses")
+ .add_entity(adalias)
+ .order_by(User.id)
+ .order_by(adalias.id)
+ .all(),
+ [
+ (
+ 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,
+ )
+
+ self.mapper_registry.map_imperatively(
+ User,
+ users,
+ properties={"orders": relationship(Order, backref="user")},
+ ) # o2m, m2o
+ self.mapper_registry.map_imperatively(
+ Order,
+ orders,
+ properties={
+ "items": relationship(
+ Item, secondary=order_items, order_by=items.c.id
+ )
+ },
+ ) # m2m
+
+ self.mapper_registry.map_imperatively(
+ Item,
+ items,
+ properties={
+ "keywords": relationship(
+ Keyword, secondary=item_keywords, order_by=keywords.c.id
+ )
+ },
+ ) # m2m
+ self.mapper_registry.map_imperatively(Keyword, keywords)
+
+ sess = fixture_session()
+ sel = users.select().where(users.c.id.in_([7, 8]))
+
+ with assertions.expect_deprecated_20(sef_dep):
+ eq_(
+ sess.query(User)
+ .select_entity_from(sel.subquery())
+ .join(User.orders, Order.items, Item.keywords)
+ .filter(Keyword.name.in_(["red", "big", "round"]))
+ .all(),
+ [User(name="jack", id=7)],
+ )
+
+ def test_very_nested_joins_with_joinedload(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,
+ )
+
+ self.mapper_registry.map_imperatively(
+ User,
+ users,
+ properties={"orders": relationship(Order, backref="user")},
+ ) # o2m, m2o
+ self.mapper_registry.map_imperatively(
+ Order,
+ orders,
+ properties={
+ "items": relationship(
+ Item, secondary=order_items, order_by=items.c.id
+ )
+ },
+ ) # m2m
+ self.mapper_registry.map_imperatively(
+ Item,
+ items,
+ properties={
+ "keywords": relationship(
+ Keyword, secondary=item_keywords, order_by=keywords.c.id
+ )
+ },
+ ) # m2m
+ self.mapper_registry.map_imperatively(Keyword, keywords)
+
+ sess = fixture_session()
+
+ sel = users.select().where(users.c.id.in_([7, 8]))
+
+ def go():
+ with assertions.expect_deprecated_20(sef_dep):
+ eq_(
+ sess.query(User)
+ .select_entity_from(sel.subquery())
+ .options(
+ joinedload("orders")
+ .joinedload("items")
+ .joinedload("keywords")
+ )
+ .join(User.orders, Order.items, Item.keywords)
+ .filter(Keyword.name.in_(["red", "big", "round"]))
+ .all(),
+ [
+ User(
+ name="jack",
+ orders=[
+ Order(
+ description="order 1",
+ items=[
+ Item(
+ description="item 1",
+ keywords=[
+ Keyword(name="red"),
+ Keyword(name="big"),
+ Keyword(name="round"),
+ ],
+ ),
+ Item(
+ description="item 2",
+ keywords=[
+ Keyword(name="red", id=2),
+ Keyword(name="small", id=5),
+ Keyword(name="square"),
+ ],
+ ),
+ Item(
+ description="item 3",
+ keywords=[
+ Keyword(name="green", id=3),
+ Keyword(name="big", id=4),
+ Keyword(name="round", id=6),
+ ],
+ ),
+ ],
+ ),
+ Order(
+ description="order 3",
+ items=[
+ Item(
+ description="item 3",
+ keywords=[
+ Keyword(name="green", id=3),
+ Keyword(name="big", id=4),
+ Keyword(name="round", id=6),
+ ],
+ ),
+ Item(
+ description="item 4",
+ keywords=[],
+ id=4,
+ ),
+ Item(
+ description="item 5",
+ keywords=[],
+ id=5,
+ ),
+ ],
+ ),
+ Order(
+ description="order 5",
+ items=[
+ Item(description="item 5", keywords=[])
+ ],
+ ),
+ ],
+ )
+ ],
+ )
+
+ self.assert_sql_count(testing.db, go, 1)
+
+ sess.expunge_all()
+ sel2 = orders.select().where(orders.c.id.in_([1, 2, 3]))
+ with assertions.expect_deprecated_20(sef_dep):
+ eq_(
+ sess.query(Order)
+ .select_entity_from(sel2.subquery())
+ .join(Order.items)
+ .join(Item.keywords)
+ .filter(Keyword.name == "red")
+ .order_by(Order.id)
+ .all(),
+ [
+ Order(description="order 1", id=1),
+ Order(description="order 2", id=2),
+ ],
+ )
+
+ def test_replace_with_eager(self):
+ users, Address, addresses, User = (
+ self.tables.users,
+ self.classes.Address,
+ self.tables.addresses,
+ self.classes.User,
+ )
+
+ self.mapper_registry.map_imperatively(
+ User,
+ users,
+ properties={
+ "addresses": relationship(Address, order_by=addresses.c.id)
+ },
+ )
+ self.mapper_registry.map_imperatively(Address, addresses)
+
+ sel = users.select().where(users.c.id.in_([7, 8]))
+ sess = fixture_session()
+
+ def go():
+ with assertions.expect_deprecated_20(sef_dep):
+ eq_(
+ sess.query(User)
+ .options(joinedload("addresses"))
+ .select_entity_from(sel.subquery())
+ .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 assertions.expect_deprecated_20(sef_dep):
+ eq_(
+ sess.query(User)
+ .options(joinedload("addresses"))
+ .select_entity_from(sel.subquery())
+ .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 assertions.expect_deprecated_20(sef_dep):
+ eq_(
+ sess.query(User)
+ .options(joinedload("addresses"))
+ .select_entity_from(sel.subquery())
+ .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)
+
+ def test_select_from_aliased_one(self):
+ User, users = self.classes.User, self.tables.users
+
+ self.mapper_registry.map_imperatively(User, users)
+
+ sess = fixture_session()
+
+ not_users = table("users", column("id"), column("name"))
+ ua = aliased(User, select(not_users).alias(), adapt_on_names=True)
+
+ with assertions.expect_deprecated_20(sef_dep):
+ q = (
+ sess.query(User.name)
+ .select_entity_from(ua)
+ .order_by(User.name)
+ )
+ self.assert_compile(
+ q,
+ "SELECT anon_1.name AS anon_1_name FROM (SELECT users.id AS id, "
+ "users.name AS name FROM users) AS anon_1 ORDER BY anon_1.name",
+ )
+ eq_(q.all(), [("chuck",), ("ed",), ("fred",), ("jack",)])
+
+ def test_select_from_aliased_two(self):
+ User, users = self.classes.User, self.tables.users
+
+ self.mapper_registry.map_imperatively(User, users)
+
+ sess = fixture_session()
+
+ ua = aliased(User)
+
+ with assertions.expect_deprecated_20(sef_dep):
+ q = (
+ sess.query(User.name)
+ .select_entity_from(ua)
+ .order_by(User.name)
+ )
+ self.assert_compile(
+ q,
+ "SELECT users_1.name AS users_1_name FROM users AS users_1 "
+ "ORDER BY users_1.name",
+ )
+ eq_(q.all(), [("chuck",), ("ed",), ("fred",), ("jack",)])
+
+ def test_select_from_core_alias_one(self):
+ User, users = self.classes.User, self.tables.users
+
+ self.mapper_registry.map_imperatively(User, users)
+
+ sess = fixture_session()
+
+ ua = users.alias()
+
+ with assertions.expect_deprecated_20(sef_dep):
+ q = (
+ sess.query(User.name)
+ .select_entity_from(ua)
+ .order_by(User.name)
+ )
+ self.assert_compile(
+ q,
+ "SELECT users_1.name AS users_1_name FROM users AS users_1 "
+ "ORDER BY users_1.name",
+ )
+ eq_(q.all(), [("chuck",), ("ed",), ("fred",), ("jack",)])
+
+ def test_differentiate_self_external(self):
+ """test some different combinations of joining a table to a subquery of
+ itself."""
+
+ users, User = self.tables.users, self.classes.User
+
+ self.mapper_registry.map_imperatively(User, users)
+
+ sess = fixture_session()
+
+ sel = sess.query(User).filter(User.id.in_([7, 8])).subquery()
+ ualias = aliased(User)
+
+ with assertions.expect_deprecated_20(sef_dep):
+ self.assert_compile(
+ sess.query(ualias)
+ .select_entity_from(sel)
+ .filter(ualias.id > sel.c.id),
+ "SELECT users_1.id AS users_1_id, "
+ "users_1.name AS users_1_name "
+ "FROM users AS users_1, ("
+ "SELECT users.id AS id, users.name AS name FROM users "
+ "WHERE users.id IN ([POSTCOMPILE_id_1])) AS anon_1 "
+ "WHERE users_1.id > anon_1.id",
+ check_post_param={"id_1": [7, 8]},
+ )
+
+ with assertions.expect_deprecated_20(sef_dep):
+ self.assert_compile(
+ sess.query(ualias)
+ .select_entity_from(sel)
+ .join(ualias, ualias.id > sel.c.id),
+ "SELECT users_1.id AS users_1_id, "
+ "users_1.name AS users_1_name "
+ "FROM (SELECT users.id AS id, users.name AS name "
+ "FROM users WHERE users.id IN ([POSTCOMPILE_id_1])) AS anon_1 "
+ "JOIN users AS users_1 ON users_1.id > anon_1.id",
+ check_post_param={"id_1": [7, 8]},
+ )
+
+ with assertions.expect_deprecated_20(sef_dep):
+ self.assert_compile(
+ sess.query(ualias)
+ .select_entity_from(sel)
+ .join(ualias, ualias.id > User.id),
+ "SELECT users_1.id AS users_1_id, "
+ "users_1.name AS users_1_name "
+ "FROM (SELECT users.id AS id, users.name AS name FROM "
+ "users WHERE users.id IN ([POSTCOMPILE_id_1])) AS anon_1 "
+ "JOIN users AS users_1 ON users_1.id > anon_1.id",
+ check_post_param={"id_1": [7, 8]},
+ )
+
+ with assertions.expect_deprecated_20(sef_dep):
+ self.assert_compile(
+ sess.query(ualias).select_entity_from(
+ join(sel, ualias, ualias.id > sel.c.id)
+ ),
+ "SELECT users_1.id AS users_1_id, "
+ "users_1.name AS users_1_name "
+ "FROM "
+ "(SELECT users.id AS id, users.name AS name "
+ "FROM users WHERE users.id "
+ "IN ([POSTCOMPILE_id_1])) AS anon_1 "
+ "JOIN users AS users_1 ON users_1.id > anon_1.id",
+ check_post_param={"id_1": [7, 8]},
+ )
+
+
+class JoinLateralTest(fixtures.MappedTest, AssertsCompiledSQL):
+ __dialect__ = default.DefaultDialect(supports_native_boolean=True)
+
+ run_setup_bind = None
+ run_setup_mappers = "once"
+
+ run_create_tables = None
+
+ @classmethod
+ def define_tables(cls, metadata):
+ Table(
+ "people",
+ metadata,
+ Column("people_id", Integer, primary_key=True),
+ Column("age", Integer),
+ Column("name", String(30)),
+ )
+ Table(
+ "bookcases",
+ metadata,
+ Column("bookcase_id", Integer, primary_key=True),
+ Column(
+ "bookcase_owner_id", Integer, ForeignKey("people.people_id")
+ ),
+ Column("bookcase_shelves", Integer),
+ Column("bookcase_width", Integer),
+ )
+ Table(
+ "books",
+ metadata,
+ Column("book_id", Integer, primary_key=True),
+ Column(
+ "bookcase_id", Integer, ForeignKey("bookcases.bookcase_id")
+ ),
+ Column("book_owner_id", Integer, ForeignKey("people.people_id")),
+ Column("book_weight", Integer),
+ )
+
+ @classmethod
+ def setup_classes(cls):
+ class Person(cls.Comparable):
+ pass
+
+ class Bookcase(cls.Comparable):
+ pass
+
+ class Book(cls.Comparable):
+ pass
+
+ @classmethod
+ def setup_mappers(cls):
+ Person, Bookcase, Book = cls.classes("Person", "Bookcase", "Book")
+ people, bookcases, books = cls.tables("people", "bookcases", "books")
+ cls.mapper_registry.map_imperatively(Person, people)
+ cls.mapper_registry.map_imperatively(
+ Bookcase,
+ bookcases,
+ properties={
+ "owner": relationship(Person),
+ "books": relationship(Book),
+ },
+ )
+ cls.mapper_registry.map_imperatively(Book, books)
+
+ # "sef" == "select entity from"
+ def test_select_subquery_sef_implicit_correlate(self):
+ Person, Book = self.classes("Person", "Book")
+
+ s = fixture_session()
+
+ stmt = s.query(Person).subquery()
+
+ subq = (
+ s.query(Book.book_id)
+ .filter(Person.people_id == Book.book_owner_id)
+ .subquery()
+ .lateral()
+ )
+
+ with assertions.expect_deprecated_20(sef_dep):
+ stmt = (
+ s.query(Person, subq.c.book_id)
+ .select_entity_from(stmt)
+ .join(subq, true())
+ )
+
+ self.assert_compile(
+ stmt,
+ "SELECT anon_1.people_id AS anon_1_people_id, "
+ "anon_1.age AS anon_1_age, anon_1.name AS anon_1_name, "
+ "anon_2.book_id AS anon_2_book_id "
+ "FROM "
+ "(SELECT people.people_id AS people_id, people.age AS age, "
+ "people.name AS name FROM people) AS anon_1 "
+ "JOIN LATERAL "
+ "(SELECT books.book_id AS book_id FROM books "
+ "WHERE anon_1.people_id = books.book_owner_id) AS anon_2 ON true",
+ )
+
+ def test_select_subquery_sef_implicit_correlate_coreonly(self):
+ Person, Book = self.classes("Person", "Book")
+
+ s = fixture_session()
+
+ stmt = s.query(Person).subquery()
+
+ subq = (
+ select(Book.book_id)
+ .where(Person.people_id == Book.book_owner_id)
+ .subquery()
+ .lateral()
+ )
+
+ with assertions.expect_deprecated_20(sef_dep):
+ stmt = (
+ s.query(Person, subq.c.book_id)
+ .select_entity_from(stmt)
+ .join(subq, true())
+ )
+
+ self.assert_compile(
+ stmt,
+ "SELECT anon_1.people_id AS anon_1_people_id, "
+ "anon_1.age AS anon_1_age, anon_1.name AS anon_1_name, "
+ "anon_2.book_id AS anon_2_book_id "
+ "FROM "
+ "(SELECT people.people_id AS people_id, people.age AS age, "
+ "people.name AS name FROM people) AS anon_1 "
+ "JOIN LATERAL "
+ "(SELECT books.book_id AS book_id FROM books "
+ "WHERE anon_1.people_id = books.book_owner_id) AS anon_2 ON true",
+ )
+
+ def test_select_subquery_sef_explicit_correlate_coreonly(self):
+ Person, Book = self.classes("Person", "Book")
+
+ s = fixture_session()
+
+ stmt = s.query(Person).subquery()
+
+ subq = (
+ select(Book.book_id)
+ .correlate(Person)
+ .where(Person.people_id == Book.book_owner_id)
+ .subquery()
+ .lateral()
+ )
+
+ with assertions.expect_deprecated_20(sef_dep):
+ stmt = (
+ s.query(Person, subq.c.book_id)
+ .select_entity_from(stmt)
+ .join(subq, true())
+ )
+
+ self.assert_compile(
+ stmt,
+ "SELECT anon_1.people_id AS anon_1_people_id, "
+ "anon_1.age AS anon_1_age, anon_1.name AS anon_1_name, "
+ "anon_2.book_id AS anon_2_book_id "
+ "FROM "
+ "(SELECT people.people_id AS people_id, people.age AS age, "
+ "people.name AS name FROM people) AS anon_1 "
+ "JOIN LATERAL "
+ "(SELECT books.book_id AS book_id FROM books "
+ "WHERE anon_1.people_id = books.book_owner_id) AS anon_2 ON true",
+ )
+
+ def test_select_subquery_sef_explicit_correlate(self):
+ Person, Book = self.classes("Person", "Book")
+
+ s = fixture_session()
+
+ stmt = s.query(Person).subquery()
+
+ subq = (
+ s.query(Book.book_id)
+ .correlate(Person)
+ .filter(Person.people_id == Book.book_owner_id)
+ .subquery()
+ .lateral()
+ )
+
+ with assertions.expect_deprecated_20(sef_dep):
+ stmt = (
+ s.query(Person, subq.c.book_id)
+ .select_entity_from(stmt)
+ .join(subq, true())
+ )
+
+ self.assert_compile(
+ stmt,
+ "SELECT anon_1.people_id AS anon_1_people_id, "
+ "anon_1.age AS anon_1_age, anon_1.name AS anon_1_name, "
+ "anon_2.book_id AS anon_2_book_id "
+ "FROM "
+ "(SELECT people.people_id AS people_id, people.age AS age, "
+ "people.name AS name FROM people) AS anon_1 "
+ "JOIN LATERAL "
+ "(SELECT books.book_id AS book_id FROM books "
+ "WHERE anon_1.people_id = books.book_owner_id) AS anon_2 ON true",
+ )
+
+ def test_from_function_sef(self):
+ Bookcase = self.classes.Bookcase
+
+ s = fixture_session()
+
+ subq = s.query(Bookcase).subquery()
+
+ srf = lateral(func.generate_series(1, Bookcase.bookcase_shelves))
+
+ with assertions.expect_deprecated_20(sef_dep):
+ q = s.query(Bookcase).select_entity_from(subq).join(srf, true())
+
+ self.assert_compile(
+ q,
+ "SELECT anon_1.bookcase_id AS anon_1_bookcase_id, "
+ "anon_1.bookcase_owner_id AS anon_1_bookcase_owner_id, "
+ "anon_1.bookcase_shelves AS anon_1_bookcase_shelves, "
+ "anon_1.bookcase_width AS anon_1_bookcase_width "
+ "FROM (SELECT bookcases.bookcase_id AS bookcase_id, "
+ "bookcases.bookcase_owner_id AS bookcase_owner_id, "
+ "bookcases.bookcase_shelves AS bookcase_shelves, "
+ "bookcases.bookcase_width AS bookcase_width FROM bookcases) "
+ "AS anon_1 "
+ "JOIN LATERAL "
+ "generate_series(:generate_series_1, anon_1.bookcase_shelves) "
+ "AS anon_2 ON true",
+ )
-import sqlalchemy as sa
from sqlalchemy import and_
from sqlalchemy import asc
from sqlalchemy import desc
from sqlalchemy.orm import relationship
from sqlalchemy.orm import Session
from sqlalchemy.orm.context import ORMSelectCompileState
-from sqlalchemy.orm.util import join
from sqlalchemy.sql import column
from sqlalchemy.sql import table
from sqlalchemy.sql.selectable import LABEL_STYLE_TABLENAME_PLUS_COL
self.assert_compile(
sess.query(users)
- .select_entity_from(users.select().subquery())
+ .select_from(users.select().subquery())
.set_label_style(LABEL_STYLE_TABLENAME_PLUS_COL)
.statement,
"SELECT users.id AS users_id, users.name AS users_name "
self.assert_compile(
sess.query(users, s.c.email)
- .select_entity_from(users.join(s, s.c.id == users.c.id))
+ .select_from(users.join(s, s.c.id == users.c.id))
.set_label_style(LABEL_STYLE_TABLENAME_PLUS_COL)
.statement,
"SELECT users.id AS users_id, users.name AS users_name, "
__dialect__ = "default"
- def test_select_entity_from(self):
- User = self.classes.User
- sess = fixture_session()
-
- q = sess.query(User)
- 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 "
- "FROM (SELECT users.id AS id, users.name AS name FROM "
- "users) AS anon_1 WHERE anon_1.name = :name_1",
- )
-
- def test_select_entity_from_no_entities(self):
- User = self.classes.User
- sess = fixture_session()
-
- assert_raises_message(
- sa.exc.ArgumentError,
- r"A selectable \(FromClause\) instance is "
- "expected when the base alias is being set",
- sess.query(User).select_entity_from(User)._compile_context,
- )
-
def test_select_from_no_aliasing(self):
User = self.classes.User
sess = fixture_session()
.order_by(text("ulist.id"), addresses.c.id)
)
sess = fixture_session()
- q = sess.query(User)
def go():
ulist = query.alias("ulist")
ulist_alias = aliased(User, alias=ulist)
- result = (
- q.options(contains_eager(User.addresses, alias=ulist))
- .select_entity_from(ulist_alias)
- .all()
- )
+ q = sess.query(ulist_alias)
+
+ result = q.options(
+ contains_eager(ulist_alias.addresses, alias=ulist)
+ ).all()
assert self.static.user_address_result == result
self.assert_sql_count(testing.db, go, 1)
)
sess = fixture_session()
- # better way. use select_entity_from()
def go():
+ qs = query.subquery()
+ ua = aliased(User, qs)
result = (
- sess.query(User)
- .select_entity_from(query.subquery())
- .options(contains_eager(User.addresses))
+ sess.query(ua)
+ .options(contains_eager(ua.addresses, alias=qs))
.all()
)
assert self.static.user_address_result == result
self.tables.addresses,
self.tables.users,
)
+ Address = self.classes.Address
sess = fixture_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()
)
def go():
+ qs = query.subquery()
+ ua = aliased(User, qs)
+ aa = aliased(Address, qs)
result = (
- sess.query(User)
- .select_entity_from(query.subquery())
- .options(contains_eager(User.addresses, alias=adalias))
+ sess.query(ua)
+ .options(contains_eager(ua.addresses.of_type(aa)))
.all()
)
assert self.static.user_address_result == result
sess = fixture_session()
adalias = addresses.alias()
+ qq = users.outerjoin(adalias)
+ ua = aliased(User, qq)
q = (
- sess.query(User)
- .select_entity_from(users.outerjoin(adalias))
- .options(contains_eager(User.addresses, alias=adalias))
+ sess.query(ua)
+ .options(contains_eager(ua.addresses, alias=adalias))
.order_by(User.id, adalias.c.id)
)
sess = fixture_session()
sel = users.select().where(User.id.in_([7, 8])).alias()
- q = sess.query(User.name)
- q2 = q.select_entity_from(sel).all()
+ ua = aliased(User, sel)
+ q = sess.query(ua.name)
+ q2 = q.all()
eq_(list(q2), [("jack",), ("ed",)])
def test_column_queries_three(self):
.limit(10)
)
- # test eager aliasing, with/without select_entity_from aliasing
for q in [q1, q2]:
eq_(
q.all(),
(user10, None),
]
- q = sess.query(User)
adalias = addresses.alias("adalias")
- q = q.add_entity(Address, alias=adalias).select_entity_from(
- users.outerjoin(adalias)
- )
+ uaj = users.outerjoin(adalias)
+ ua = aliased(User, uaj)
+ q = sess.query(ua)
+ q = q.add_entity(Address, alias=adalias)
result = q.order_by(User.id, adalias.c.id).all()
assert result == expected
sess.expunge_all()
- q = sess.query(User).add_entity(Address, alias=adalias)
- result = (
- q.select_entity_from(users.outerjoin(adalias))
- .filter(adalias.c.email_address == "ed@bettyboop.com")
- .all()
- )
+ q = sess.query(ua).add_entity(Address, alias=adalias)
+ result = q.filter(adalias.c.email_address == "ed@bettyboop.com").all()
assert result == [(user8, address3)]
def test_with_entities(self):
eq_(result, expected)
with fixture_session() as sess:
- # test with select_entity_from()
+ uaj = users.outerjoin(addresses)
+ ua = aliased(User, uaj)
q = (
fixture_session()
- .query(User)
+ .query(ua)
.add_columns(
func.count(addresses.c.id), ("Name:" + users.c.name)
)
- .select_entity_from(users.outerjoin(addresses))
.group_by(users)
.order_by(users.c.id)
)
sel = users.select().where(users.c.id.in_([7, 8])).alias()
sess = fixture_session()
+ ua = aliased(User, sel)
eq_(
- sess.query(User).select_entity_from(sel).all(),
+ sess.query(ua).all(),
[User(id=7), User(id=8)],
)
eq_(
- sess.query(User)
- .select_entity_from(sel)
- .filter(User.id == 8)
- .all(),
+ sess.query(ua).filter(ua.id == 8).all(),
[User(id=8)],
)
eq_(
- sess.query(User)
- .select_entity_from(sel)
- .order_by(desc(User.name))
- .all(),
+ sess.query(ua).order_by(desc(ua.name)).all(),
[User(name="jack", id=7), User(name="ed", id=8)],
)
eq_(
- sess.query(User)
- .select_entity_from(sel)
- .order_by(asc(User.name))
- .all(),
+ sess.query(ua).order_by(asc(ua.name)).all(),
[User(name="ed", id=8), User(name="jack", id=7)],
)
eq_(
- sess.query(User)
- .select_entity_from(sel)
- .options(joinedload(User.addresses))
- .first(),
+ sess.query(ua).options(joinedload(ua.addresses)).first(),
User(name="jack", addresses=[Address(id=1)]),
)
not_users = table("users", column("id"), column("name"))
ua = aliased(User, select(not_users).alias(), adapt_on_names=True)
- q = sess.query(User.name).select_entity_from(ua).order_by(User.name)
+ q = sess.query(ua.name).order_by(ua.name)
self.assert_compile(
q,
"SELECT anon_1.name AS anon_1_name FROM (SELECT users.id AS id, "
ua = aliased(User)
- q = sess.query(User.name).select_entity_from(ua).order_by(User.name)
- self.assert_compile(
- q,
- "SELECT users_1.name AS users_1_name FROM users AS users_1 "
- "ORDER BY users_1.name",
- )
- eq_(q.all(), [("chuck",), ("ed",), ("fred",), ("jack",)])
-
- def test_select_from_core_alias_one(self):
- User, users = self.classes.User, self.tables.users
-
- self.mapper_registry.map_imperatively(User, users)
-
- sess = fixture_session()
-
- ua = users.alias()
-
- q = sess.query(User.name).select_entity_from(ua).order_by(User.name)
+ q = sess.query(ua.name).order_by(ua.name)
self.assert_compile(
q,
"SELECT users_1.name AS users_1_name FROM users AS users_1 "
sel = sess.query(User).filter(User.id.in_([7, 8])).subquery()
ualias = aliased(User)
+ ua = aliased(User, sel)
+
self.assert_compile(
sess.query(User).join(sel, User.id > sel.c.id),
"SELECT users.id AS users_id, users.name AS users_name FROM "
)
self.assert_compile(
- sess.query(ualias)
- .select_entity_from(sel)
- .filter(ualias.id > sel.c.id),
+ sess.query(ualias).select_from(ua).filter(ualias.id > ua.id),
"SELECT users_1.id AS users_1_id, users_1.name AS users_1_name "
"FROM users AS users_1, ("
"SELECT users.id AS id, users.name AS name FROM users "
)
self.assert_compile(
- sess.query(ualias)
- .select_entity_from(sel)
- .join(ualias, ualias.id > sel.c.id),
+ sess.query(ualias).select_from(ua).join(ualias, ualias.id > ua.id),
"SELECT users_1.id AS users_1_id, users_1.name AS users_1_name "
"FROM (SELECT users.id AS id, users.name AS name "
"FROM users WHERE users.id IN ([POSTCOMPILE_id_1])) AS anon_1 "
)
self.assert_compile(
- sess.query(ualias)
- .select_entity_from(sel)
- .join(ualias, ualias.id > User.id),
+ sess.query(ualias).select_from(ua).join(ualias, ualias.id > ua.id),
"SELECT users_1.id AS users_1_id, users_1.name AS users_1_name "
"FROM (SELECT users.id AS id, users.name AS name FROM "
"users WHERE users.id IN ([POSTCOMPILE_id_1])) AS anon_1 "
check_post_param={"id_1": [7, 8]},
)
- self.assert_compile(
- sess.query(ualias).select_entity_from(
- join(sel, ualias, ualias.id > sel.c.id)
- ),
- "SELECT users_1.id AS users_1_id, users_1.name AS users_1_name "
- "FROM "
- "(SELECT users.id AS id, users.name AS name "
- "FROM users WHERE users.id "
- "IN ([POSTCOMPILE_id_1])) AS anon_1 "
- "JOIN users AS users_1 ON users_1.id > anon_1.id",
- check_post_param={"id_1": [7, 8]},
- )
-
def test_aliased_class_vs_nonaliased(self):
User, users = self.classes.User, self.tables.users
self.mapper_registry.map_imperatively(User, users)
"FROM users JOIN users AS users_1 ON users.name > users_1.name",
)
- # this is tested in many other places here, just adding it
- # here for comparison
- self.assert_compile(
- sess.query(User.name).select_entity_from(
- 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",
- )
-
- def test_join_no_order_by(self):
- User, users = self.classes.User, self.tables.users
-
- self.mapper_registry.map_imperatively(User, users)
-
- sel = users.select().where(users.c.id.in_([7, 8]))
- sess = fixture_session()
-
- eq_(
- sess.query(User).select_entity_from(sel.subquery()).all(),
- [User(name="jack", id=7), User(name="ed", id=8)],
- )
-
def test_join_relname_from_selected_from(self):
User, Address = self.classes.User, self.classes.Address
users, addresses = self.tables.users, self.tables.addresses
sel = users.select().where(users.c.id.in_([7, 8]))
sess = fixture_session()
+ ua = aliased(User, sel.subquery())
eq_(
- sess.query(User)
- .select_entity_from(sel.subquery())
- .join(User.addresses)
+ sess.query(ua)
+ .join(ua.addresses)
.add_entity(Address)
- .order_by(User.id)
+ .order_by(ua.id)
.order_by(Address.id)
.all(),
[
)
adalias = aliased(Address)
+ ua = aliased(User, sel.subquery())
eq_(
- sess.query(User)
- .select_entity_from(sel.subquery())
- .join(adalias, User.addresses)
+ sess.query(ua)
+ .join(ua.addresses.of_type(adalias))
.add_entity(adalias)
- .order_by(User.id)
+ .order_by(ua.id)
.order_by(adalias.id)
.all(),
[
sess = fixture_session()
sel = users.select().where(users.c.id.in_([7, 8]))
+ ua = aliased(User, sel.subquery())
eq_(
- sess.query(User)
- .select_entity_from(sel.subquery())
- .join(User.orders)
+ sess.query(ua)
+ .join(ua.orders)
.join(Order.items)
.join(Item.keywords)
.filter(Keyword.name.in_(["red", "big", "round"]))
sess = fixture_session()
sel = users.select().where(users.c.id.in_([7, 8]))
+ ua = aliased(User, sel.subquery())
def go():
eq_(
- sess.query(User)
- .select_entity_from(sel.subquery())
+ sess.query(ua)
.options(
- joinedload(User.orders)
+ joinedload(ua.orders)
.joinedload(Order.items)
.joinedload(Item.keywords)
)
- .join(User.orders)
+ .join(ua.orders)
.join(Order.items)
.join(Item.keywords)
.filter(Keyword.name.in_(["red", "big", "round"]))
sess.expunge_all()
sel2 = orders.select().where(orders.c.id.in_([1, 2, 3]))
+ oa = aliased(Order, sel2.subquery())
eq_(
- sess.query(Order)
- .select_entity_from(sel2.subquery())
- .join(Order.items)
+ sess.query(oa)
+ .join(oa.items)
.join(Item.keywords)
.filter(Keyword.name == "red")
- .order_by(Order.id)
+ .order_by(oa.id)
.all(),
[
Order(description="order 1", id=1),
sel = users.select().where(users.c.id.in_([7, 8]))
sess = fixture_session()
+ ua = aliased(User, sel.subquery())
+
def go():
eq_(
- sess.query(User)
- .options(joinedload(User.addresses))
- .select_entity_from(sel.subquery())
- .order_by(User.id)
+ sess.query(ua)
+ .options(joinedload(ua.addresses))
+ .order_by(ua.id)
.all(),
[
User(id=7, addresses=[Address(id=1)]),
def go():
eq_(
- sess.query(User)
- .options(joinedload(User.addresses))
- .select_entity_from(sel.subquery())
- .filter(User.id == 8)
- .order_by(User.id)
+ sess.query(ua)
+ .options(joinedload(ua.addresses))
+ .filter(ua.id == 8)
+ .order_by(ua.id)
.all(),
[
User(
def go():
eq_(
- sess.query(User)
- .options(joinedload(User.addresses))
- .select_entity_from(sel.subquery())
- .order_by(User.id)[1],
+ sess.query(ua)
+ .options(joinedload(ua.addresses))
+ .order_by(ua.id)
+ .offset(1)
+ .limit(1)
+ .scalar(),
User(
id=8,
addresses=[Address(id=2), Address(id=3), Address(id=4)],
Address(id=4, user=User(id=8, concat=16, count=3)),
Address(id=5, user=User(id=9, concat=18, count=1)),
]
- # TODO: ISSUE: BUG: cached metadata is confusing the user.id
- # column here with the anon_1 for some reason, when we
- # use compiled cache. this bug may even be present in
- # regular main / 1.3. right now the caching of result
- # metadata is disabled.
+
eq_(sess.query(Address).all(), address_result)
# run the eager version twice to test caching of aliased clauses
ua = aliased(User)
eq_(
sess.query(Address, ua.concat, ua.count)
- .select_entity_from(join(Address, ua, "user"))
+ .join(Address.user.of_type(ua))
.options(joinedload(Address.user))
.order_by(Address.id)
.all(),
eq_(
list(
sess.query(Address, ua)
- .select_entity_from(join(Address, ua, Address.user))
+ .join(Address.user.of_type(ua))
.with_entities(Address.id, ua.id, ua.concat, ua.count)
),
[
)
def test_clause_onclause(self):
- Item, Order, users, order_items, User = (
+ Item, Order, order_items, User = (
self.classes.Item,
self.classes.Order,
- self.tables.users,
self.tables.order_items,
self.classes.User,
)
[User(name="fred")],
)
- # same with an explicit select_from()
- eq_(
- sess.query(User)
- .select_entity_from(
- select(users).order_by(User.id).offset(2).alias()
- )
- .join(Order, User.id == Order.user_id)
- .all(),
- [User(name="fred")],
- )
-
def test_aliased_classes(self):
User, Address = self.classes.User, self.classes.Address
"WHERE people.people_id = books.book_owner_id) AS anon_1 ON true",
)
- # sef == select_entity_from
- def test_select_subquery_sef_implicit_correlate(self):
+ # "aas" == "aliased against select"
+ def test_select_subquery_aas_implicit_correlate(self):
Person, Book = self.classes("Person", "Book")
s = fixture_session()
stmt = s.query(Person).subquery()
+ pa = aliased(Person, stmt)
+
subq = (
s.query(Book.book_id)
- .filter(Person.people_id == Book.book_owner_id)
+ .filter(pa.people_id == Book.book_owner_id)
.subquery()
.lateral()
)
- stmt = (
- s.query(Person, subq.c.book_id)
- .select_entity_from(stmt)
- .join(subq, true())
- )
+ stmt = s.query(pa, subq.c.book_id).join(subq, true())
self.assert_compile(
stmt,
"WHERE anon_1.people_id = books.book_owner_id) AS anon_2 ON true",
)
- def test_select_subquery_sef_implicit_correlate_coreonly(self):
+ def test_select_subquery_aas_implicit_correlate_coreonly(self):
Person, Book = self.classes("Person", "Book")
s = fixture_session()
stmt = s.query(Person).subquery()
+ pa = aliased(Person, stmt)
+
subq = (
select(Book.book_id)
- .where(Person.people_id == Book.book_owner_id)
+ .where(pa.people_id == Book.book_owner_id)
.subquery()
.lateral()
)
- stmt = (
- s.query(Person, subq.c.book_id)
- .select_entity_from(stmt)
- .join(subq, true())
- )
+ stmt = s.query(pa, subq.c.book_id).join(subq, true())
self.assert_compile(
stmt,
"WHERE anon_1.people_id = books.book_owner_id) AS anon_2 ON true",
)
- def test_select_subquery_sef_explicit_correlate_coreonly(self):
+ def test_select_subquery_aas_explicit_correlate_coreonly(self):
Person, Book = self.classes("Person", "Book")
s = fixture_session()
stmt = s.query(Person).subquery()
+ pa = aliased(Person, stmt)
+
subq = (
select(Book.book_id)
- .correlate(Person)
- .where(Person.people_id == Book.book_owner_id)
+ .correlate(pa)
+ .where(pa.people_id == Book.book_owner_id)
.subquery()
.lateral()
)
- stmt = (
- s.query(Person, subq.c.book_id)
- .select_entity_from(stmt)
- .join(subq, true())
- )
+ stmt = s.query(pa, subq.c.book_id).join(subq, true())
self.assert_compile(
stmt,
"WHERE anon_1.people_id = books.book_owner_id) AS anon_2 ON true",
)
- def test_select_subquery_sef_explicit_correlate(self):
+ def test_select_subquery_aas_explicit_correlate(self):
Person, Book = self.classes("Person", "Book")
s = fixture_session()
stmt = s.query(Person).subquery()
+ pa = aliased(Person, stmt)
subq = (
s.query(Book.book_id)
- .correlate(Person)
- .filter(Person.people_id == Book.book_owner_id)
+ .correlate(pa)
+ .filter(pa.people_id == Book.book_owner_id)
.subquery()
.lateral()
)
- stmt = (
- s.query(Person, subq.c.book_id)
- .select_entity_from(stmt)
- .join(subq, true())
- )
+ stmt = s.query(pa, subq.c.book_id).join(subq, true())
self.assert_compile(
stmt,
"bookcases.bookcase_shelves) AS anon_1 ON true",
)
- def test_from_function_select_entity_from(self):
+ def test_from_function_aas(self):
Bookcase = self.classes.Bookcase
s = fixture_session()
subq = s.query(Bookcase).subquery()
- srf = lateral(func.generate_series(1, Bookcase.bookcase_shelves))
+ ba = aliased(Bookcase, subq)
+
+ srf = lateral(func.generate_series(1, ba.bookcase_shelves))
self.assert_compile(
- s.query(Bookcase).select_entity_from(subq).join(srf, true()),
+ s.query(ba).join(srf, true()),
"SELECT anon_1.bookcase_id AS anon_1_bookcase_id, "
"anon_1.bookcase_owner_id AS anon_1_bookcase_owner_id, "
"anon_1.bookcase_shelves AS anon_1_bookcase_shelves, "
[User(id=7), User(id=8), User(id=9), User(id=10)],
)
- def test_via_textasfrom_select_from(self):
+ def test_via_textasfrom_aliased(self):
User = self.classes.User
s = fixture_session()
+ ua = aliased(
+ User,
+ text("select * from users").columns(User.id, User.name).subquery(),
+ )
+
eq_(
- s.query(User)
- .select_entity_from(
- text("select * from users")
- .columns(User.id, User.name)
- .subquery()
- )
- .order_by(User.id)
- .all(),
+ s.query(ua).order_by(ua.id).all(),
[User(id=7), User(id=8), User(id=9), User(id=10)],
)
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
