import itertools
import sqlalchemy as sa
from sqlalchemy import and_
from sqlalchemy import desc
from sqlalchemy import exc as sa_exc
from sqlalchemy import ForeignKey
from sqlalchemy import func
from sqlalchemy import inspect
from sqlalchemy import Integer
from sqlalchemy import lateral
from sqlalchemy import literal_column
from sqlalchemy import MetaData
from sqlalchemy import not_
from sqlalchemy import or_
from sqlalchemy import select
from sqlalchemy import String
from sqlalchemy import Table
from sqlalchemy import testing
from sqlalchemy import true
from sqlalchemy import union
from sqlalchemy.engine import default
from sqlalchemy.orm import aliased
from sqlalchemy.orm import backref
from sqlalchemy.orm import join
from sqlalchemy.orm import joinedload
from sqlalchemy.orm import outerjoin
from sqlalchemy.orm import relationship
from sqlalchemy.orm import Session
from sqlalchemy.orm import synonym
from sqlalchemy.sql.selectable import LABEL_STYLE_TABLENAME_PLUS_COL
from sqlalchemy.testing import assert_raises
from sqlalchemy.testing import assert_raises_message
from sqlalchemy.testing import AssertsCompiledSQL
from sqlalchemy.testing import eq_
from sqlalchemy.testing import fixtures
from sqlalchemy.testing.assertions import expect_raises_message
from sqlalchemy.testing.fixtures import fixture_session
from sqlalchemy.testing.schema import Column
from test.orm import _fixtures
from .inheritance import _poly_fixtures
from .test_query import QueryTest
class InheritedTest(_poly_fixtures._Polymorphic):
run_setup_mappers = "once"
class InheritedJoinTest(InheritedTest, AssertsCompiledSQL):
def test_single_prop(self):
Company = self.classes.Company
sess = fixture_session()
self.assert_compile(
sess.query(Company).join(Company.employees),
"SELECT companies.company_id AS companies_company_id, "
"companies.name AS companies_name "
"FROM companies JOIN people "
"ON companies.company_id = people.company_id",
use_default_dialect=True,
)
def test_force_via_select_from(self):
Company, Engineer = self.classes.Company, self.classes.Engineer
sess = fixture_session()
self.assert_compile(
sess.query(Company)
.filter(Company.company_id == Engineer.company_id)
.filter(Engineer.primary_language == "java"),
"SELECT companies.company_id AS companies_company_id, "
"companies.name AS companies_name "
"FROM companies, people, engineers "
"WHERE companies.company_id = people.company_id "
"AND engineers.primary_language "
"= :primary_language_1",
use_default_dialect=True,
)
self.assert_compile(
sess.query(Company)
.select_from(Company, Engineer)
.filter(Company.company_id == Engineer.company_id)
.filter(Engineer.primary_language == "java"),
"SELECT companies.company_id AS companies_company_id, "
"companies.name AS companies_name "
"FROM companies, people JOIN engineers "
"ON people.person_id = engineers.person_id "
"WHERE companies.company_id = people.company_id "
"AND engineers.primary_language ="
" :primary_language_1",
use_default_dialect=True,
)
def test_single_prop_of_type(self):
Company, Engineer = self.classes.Company, self.classes.Engineer
sess = fixture_session()
self.assert_compile(
sess.query(Company).join(Company.employees.of_type(Engineer)),
"SELECT companies.company_id AS companies_company_id, "
"companies.name AS companies_name "
"FROM companies JOIN "
"(people JOIN engineers "
"ON people.person_id = engineers.person_id) "
"ON companies.company_id = people.company_id",
use_default_dialect=True,
)
def test_explicit_polymorphic_join_one(self):
Company, Engineer = self.classes.Company, self.classes.Engineer
sess = fixture_session()
self.assert_compile(
sess.query(Company)
.join(Engineer)
.filter(Engineer.engineer_name == "vlad"),
"SELECT companies.company_id AS companies_company_id, "
"companies.name AS companies_name "
"FROM companies JOIN (people JOIN engineers "
"ON people.person_id = engineers.person_id) "
"ON "
"companies.company_id = people.company_id "
"WHERE engineers.engineer_name = :engineer_name_1",
use_default_dialect=True,
)
def test_explicit_polymorphic_join_two(self):
Company, Engineer = self.classes.Company, self.classes.Engineer
sess = fixture_session()
self.assert_compile(
sess.query(Company)
.join(Engineer, Company.company_id == Engineer.company_id)
.filter(Engineer.engineer_name == "vlad"),
"SELECT companies.company_id AS companies_company_id, "
"companies.name AS companies_name "
"FROM companies JOIN "
"(people JOIN engineers "
"ON people.person_id = engineers.person_id) "
"ON "
"companies.company_id = people.company_id "
"WHERE engineers.engineer_name = :engineer_name_1",
use_default_dialect=True,
)
def test_auto_aliasing_multi_link(self):
sess = fixture_session()
Company, Engineer, Manager, Boss = (
self.classes.Company,
self.classes.Engineer,
self.classes.Manager,
self.classes.Boss,
)
q = (
sess.query(Company)
.join(Company.employees.of_type(Engineer))
.join(Company.employees.of_type(Manager))
.join(Company.employees.of_type(Boss))
)
with testing.expect_warnings(
"An alias is being generated automatically against joined entity "
r"Mapper\[Manager\(managers\)\] due to overlapping",
"An alias is being generated automatically against joined entity "
r"Mapper\[Boss\(boss\)\] due to overlapping",
):
self.assert_compile(
q,
"SELECT companies.company_id AS companies_company_id, "
"companies.name AS companies_name FROM companies "
"JOIN (people JOIN engineers "
"ON people.person_id = engineers.person_id) "
"ON companies.company_id = people.company_id "
"JOIN (people AS people_1 JOIN managers AS managers_1 "
"ON people_1.person_id = managers_1.person_id) "
"ON companies.company_id = people_1.company_id "
"JOIN (people AS people_2 JOIN managers AS managers_2 "
"ON people_2.person_id = managers_2.person_id "
"JOIN boss AS boss_1 "
"ON managers_2.person_id = boss_1.boss_id) "
"ON companies.company_id = people_2.company_id",
use_default_dialect=True,
)
class JoinOnSynonymTest(_fixtures.FixtureTest, AssertsCompiledSQL):
__dialect__ = "default"
@classmethod
def setup_mappers(cls):
User = cls.classes.User
Address = cls.classes.Address
users, addresses = (cls.tables.users, cls.tables.addresses)
cls.mapper_registry.map_imperatively(
User,
users,
properties={
"addresses": relationship(Address),
"ad_syn": synonym("addresses"),
},
)
cls.mapper_registry.map_imperatively(Address, addresses)
def test_join_on_synonym(self):
User = self.classes.User
self.assert_compile(
fixture_session().query(User).join(User.ad_syn),
"SELECT users.id AS users_id, users.name AS users_name "
"FROM users JOIN addresses ON users.id = addresses.user_id",
)
class JoinTest(QueryTest, AssertsCompiledSQL):
__dialect__ = "default"
@testing.combinations_list(
set(
itertools.product(
[
"relationship",
"relationship_only",
"none",
"explicit",
"table_none",
"table_explicit",
],
[True, False],
)
),
argnames="onclause_type, use_legacy",
)
def test_filter_by_from_join(self, onclause_type, use_legacy):
User, Address = self.classes("User", "Address")
(address_table,) = self.tables("addresses")
(user_table,) = self.tables("users")
if use_legacy:
sess = fixture_session()
q = sess.query(User)
else:
q = select(User).set_label_style(LABEL_STYLE_TABLENAME_PLUS_COL)
if onclause_type == "relationship":
q = q.join(Address, User.addresses)
elif onclause_type == "relationship_only":
q = q.join(User.addresses)
elif onclause_type == "none":
q = q.join(Address)
elif onclause_type == "explicit":
q = q.join(Address, User.id == Address.user_id)
elif onclause_type == "table_none":
q = q.join(address_table)
elif onclause_type == "table_explicit":
q = q.join(
address_table, user_table.c.id == address_table.c.user_id
)
else:
assert False
q2 = q.filter_by(email_address="foo")
self.assert_compile(
q2,
"SELECT users.id AS users_id, users.name AS users_name "
"FROM users JOIN addresses ON users.id = addresses.user_id "
"WHERE addresses.email_address = :email_address_1",
)
if use_legacy:
q2 = q.reset_joinpoint().filter_by(name="user")
self.assert_compile(
q2,
"SELECT users.id AS users_id, users.name AS users_name "
"FROM users JOIN addresses ON users.id = addresses.user_id "
"WHERE users.name = :name_1",
)
def test_join_relationship_propagate_attrs(self):
"""test #6558"""
User = self.classes.User
users = self.tables.users
stmt = select(users).join(User.addresses)
eq_(
stmt._propagate_attrs,
{"compile_state_plugin": "orm", "plugin_subject": inspect(User)},
)
self.assert_compile(
stmt,
"SELECT users.id, users.name FROM users "
"JOIN addresses ON users.id = addresses.user_id",
)
@testing.combinations((True,), (False,), argnames="legacy")
@testing.combinations((True,), (False,), argnames="threelevel")
def test_join_with_entities(self, legacy, threelevel):
"""test issue #6503"""
User, Address, Dingaling = self.classes("User", "Address", "Dingaling")
if legacy:
sess = fixture_session()
stmt = sess.query(User).join(Address).with_entities(Address.id)
else:
stmt = select(User).join(Address).with_only_columns(Address.id)
stmt = stmt.set_label_style(LABEL_STYLE_TABLENAME_PLUS_COL)
if threelevel:
if legacy:
stmt = stmt.join(Address.dingaling).with_entities(Dingaling.id)
else:
stmt = stmt.join(Address.dingaling).with_only_columns(
Dingaling.id
)
if threelevel:
self.assert_compile(
stmt,
"SELECT dingalings.id AS dingalings_id "
"FROM users JOIN addresses ON users.id = addresses.user_id "
"JOIN dingalings ON addresses.id = dingalings.address_id",
)
else:
self.assert_compile(
stmt,
"SELECT addresses.id AS addresses_id FROM users "
"JOIN addresses ON users.id = addresses.user_id",
)
@testing.combinations((True,), (False,), argnames="legacy")
@testing.combinations((True,), (False,), argnames="threelevel")
def test_join_and_union_with_entities(self, legacy, threelevel):
"""test issue #6698, regression caused by #6503"""
User, Address, Dingaling = self.classes("User", "Address", "Dingaling")
if legacy:
sess = fixture_session()
stmt = sess.query(User).join(Address).with_entities(Address.id)
else:
stmt = select(User).join(Address).with_only_columns(Address.id)
stmt = stmt.set_label_style(LABEL_STYLE_TABLENAME_PLUS_COL)
if threelevel:
if legacy:
stmt = stmt.join(Address.dingaling).with_entities(Dingaling.id)
to_union = sess.query(Dingaling.id)
else:
stmt = stmt.join(Address.dingaling).with_only_columns(
Dingaling.id
)
to_union = select(Dingaling.id).set_label_style(
LABEL_STYLE_TABLENAME_PLUS_COL
)
else:
if legacy:
to_union = sess.query(Address.id)
else:
to_union = select(Address.id).set_label_style(
LABEL_STYLE_TABLENAME_PLUS_COL
)
if legacy:
stmt = stmt.union(to_union)
else:
stmt = (
union(stmt, to_union)
.subquery()
.select()
.set_label_style(LABEL_STYLE_TABLENAME_PLUS_COL)
)
if threelevel:
self.assert_compile(
stmt,
"SELECT anon_1.dingalings_id AS anon_1_dingalings_id FROM "
"(SELECT dingalings.id AS dingalings_id "
"FROM users JOIN addresses ON users.id = addresses.user_id "
"JOIN dingalings ON addresses.id = dingalings.address_id "
"UNION "
"SELECT dingalings.id AS dingalings_id FROM dingalings) "
"AS anon_1",
)
else:
self.assert_compile(
stmt,
"SELECT anon_1.addresses_id AS anon_1_addresses_id FROM "
"(SELECT addresses.id AS addresses_id FROM users "
"JOIN addresses ON users.id = addresses.user_id "
"UNION "
"SELECT addresses.id AS addresses_id FROM addresses) "
"AS anon_1",
)
def test_invalid_kwarg_join(self):
User = self.classes.User
sess = fixture_session()
assert_raises_message(
TypeError,
r".*join\(\) .*unexpected .*keyword",
sess.query(User).join,
"address",
foob="bar",
bar="bat",
)
assert_raises_message(
TypeError,
r".*outerjoin\(\) .*unexpected .*keyword",
sess.query(User).outerjoin,
"address",
foob="bar",
bar="bat",
)
def test_left_w_no_entity(self):
User = self.classes.User
Address = self.classes.Address
sess = fixture_session()
self.assert_compile(
sess.query(User, literal_column("x")).join(Address),
"SELECT users.id AS users_id, users.name AS users_name, x "
"FROM users JOIN addresses ON users.id = addresses.user_id",
)
self.assert_compile(
sess.query(literal_column("x"), User).join(Address),
"SELECT x, users.id AS users_id, users.name AS users_name "
"FROM users JOIN addresses ON users.id = addresses.user_id",
)
def test_left_is_none_and_query_has_no_entities(self):
Address = self.classes.Address
sess = fixture_session()
assert_raises_message(
sa_exc.InvalidRequestError,
r"No entities to join from; please use select_from\(\) to "
r"establish the left entity/selectable of this join",
sess.query().join(Address)._compile_context,
)
def test_isouter_flag(self):
User = self.classes.User
self.assert_compile(
fixture_session().query(User).join(User.orders, isouter=True),
"SELECT users.id AS users_id, users.name AS users_name "
"FROM users LEFT OUTER JOIN orders ON users.id = orders.user_id",
)
def test_full_flag(self):
User = self.classes.User
self.assert_compile(
fixture_session().query(User).outerjoin(User.orders, full=True),
"SELECT users.id AS users_id, users.name AS users_name "
"FROM users FULL OUTER JOIN orders ON users.id = orders.user_id",
)
def test_single_prop_1(self):
User = self.classes.User
sess = fixture_session()
self.assert_compile(
sess.query(User).join(User.orders),
"SELECT users.id AS users_id, users.name AS users_name "
"FROM users JOIN orders ON users.id = orders.user_id",
)
def test_single_prop_2(self):
Order, User = (self.classes.Order, self.classes.User)
sess = fixture_session()
self.assert_compile(
sess.query(User).join(Order.user),
"SELECT users.id AS users_id, users.name AS users_name "
"FROM orders JOIN users ON users.id = orders.user_id",
)
def test_single_prop_3(self):
Order, User = (self.classes.Order, self.classes.User)
sess = fixture_session()
oalias1 = aliased(Order)
self.assert_compile(
sess.query(User).join(oalias1.user),
"SELECT users.id AS users_id, users.name AS users_name "
"FROM orders AS orders_1 JOIN users "
"ON users.id = orders_1.user_id",
)
def test_single_prop_4(self):
(
Order,
User,
) = (self.classes.Order, self.classes.User)
sess = fixture_session()
oalias1 = aliased(Order)
oalias2 = aliased(Order)
self.assert_compile(
sess.query(User).join(oalias1.user).join(oalias2.user),
"SELECT users.id AS users_id, users.name AS users_name "
"FROM orders AS orders_1 JOIN users "
"ON users.id = orders_1.user_id, "
"orders AS orders_2 JOIN users ON users.id = orders_2.user_id",
)
def test_single_prop_6(self):
User = self.classes.User
sess = fixture_session()
ualias = aliased(User)
self.assert_compile(
sess.query(ualias).join(ualias.orders),
"SELECT users_1.id AS users_1_id, users_1.name AS users_1_name "
"FROM users AS users_1 JOIN orders ON users_1.id = orders.user_id",
)
def test_single_prop_9(self):
User = self.classes.User
sess = fixture_session()
subq = (
sess.query(User)
.filter(User.name == "ed")
.set_label_style(LABEL_STYLE_TABLENAME_PLUS_COL)
.subquery()
)
ua = aliased(User, subq)
self.assert_compile(
sess.query(ua).join(ua.orders),
"SELECT anon_1.users_id AS anon_1_users_id, "
"anon_1.users_name AS anon_1_users_name "
"FROM (SELECT users.id AS users_id, users.name AS users_name "
"FROM users "
"WHERE users.name = :name_1) AS anon_1 JOIN orders "
"ON anon_1.users_id = orders.user_id",
)
def test_single_prop_12(self):
Order, User, Address = (
self.classes.Order,
self.classes.User,
self.classes.Address,
)
sess = fixture_session()
oalias1 = aliased(Order)
ualias = aliased(User)
self.assert_compile(
sess.query(ualias)
.join(oalias1, ualias.orders)
.join(Address, ualias.addresses),
"SELECT users_1.id AS users_1_id, users_1.name AS "
"users_1_name FROM users AS users_1 JOIN orders AS orders_1 "
"ON users_1.id = orders_1.user_id JOIN addresses ON users_1.id "
"= addresses.user_id",
)
def test_single_prop_13(self):
Order, User, Address = (
self.classes.Order,
self.classes.User,
self.classes.Address,
)
sess = fixture_session()
ualias = aliased(User)
ualias2 = aliased(User)
self.assert_compile(
sess.query(ualias)
.join(Address, ualias.addresses)
.join(ualias2, Address.user)
.join(Order, ualias.orders),
"SELECT users_1.id AS users_1_id, users_1.name AS users_1_name "
"FROM users "
"AS users_1 JOIN addresses ON users_1.id = addresses.user_id "
"JOIN users AS users_2 "
"ON users_2.id = addresses.user_id JOIN orders "
"ON users_1.id = orders.user_id",
)
def test_overlapping_paths_one_legacy(self):
User = self.classes.User
Order = self.classes.Order
sess = fixture_session()
self.assert_compile(
sess.query(User)
.join(User.orders)
.join(Order.items)
.join(User.orders)
.join(Order.address),
"SELECT users.id AS users_id, users.name AS users_name FROM users "
"JOIN orders "
"ON users.id = orders.user_id "
"JOIN order_items AS order_items_1 "
"ON orders.id = order_items_1.order_id "
"JOIN items ON items.id = order_items_1.item_id JOIN addresses "
"ON addresses.id = orders.address_id",
)
def test_overlapping_paths_multilevel_legacy(self):
User = self.classes.User
Order = self.classes.Order
Address = self.classes.Address
s = fixture_session()
q = (
s.query(User)
.join(User.orders)
.join(User.addresses)
.join(User.orders)
.join(Order.items)
.join(User.addresses)
.join(Address.dingaling)
)
self.assert_compile(
q,
"SELECT users.id AS users_id, users.name AS users_name "
"FROM users JOIN orders ON users.id = orders.user_id "
"JOIN addresses ON users.id = addresses.user_id "
"JOIN order_items AS order_items_1 ON orders.id = "
"order_items_1.order_id "
"JOIN items ON items.id = order_items_1.item_id "
"JOIN dingalings ON addresses.id = dingalings.address_id",
)
def test_overlapping_paths_one_modern(self):
User = self.classes.User
Order = self.classes.Order
self.assert_compile(
select(User)
.join(User.orders)
.join(Order.items)
.join(User.orders)
.join(Order.address)
.set_label_style(LABEL_STYLE_TABLENAME_PLUS_COL),
"SELECT users.id AS users_id, users.name AS users_name FROM users "
"JOIN orders "
"ON users.id = orders.user_id "
"JOIN order_items AS order_items_1 "
"ON orders.id = order_items_1.order_id "
"JOIN items ON items.id = order_items_1.item_id JOIN addresses "
"ON addresses.id = orders.address_id",
)
def test_overlapping_paths_multilevel_modern(self):
User = self.classes.User
Order = self.classes.Order
Address = self.classes.Address
q = (
select(User)
.join(User.orders)
.join(User.addresses)
.join(User.orders)
.join(Order.items)
.join(User.addresses)
.join(Address.dingaling)
.set_label_style(LABEL_STYLE_TABLENAME_PLUS_COL)
)
self.assert_compile(
q,
"SELECT users.id AS users_id, users.name AS users_name "
"FROM users JOIN orders ON users.id = orders.user_id "
"JOIN addresses ON users.id = addresses.user_id "
"JOIN order_items AS order_items_1 ON orders.id = "
"order_items_1.order_id "
"JOIN items ON items.id = order_items_1.item_id "
"JOIN dingalings ON addresses.id = dingalings.address_id",
)
def test_join_nonmapped_column(self):
"""test that the search for a 'left' doesn't trip on non-mapped cols"""
Order, User = self.classes.Order, self.classes.User
sess = fixture_session()
self.assert_compile(
sess.query(User.id, literal_column("foo")).join(Order.user),
"SELECT users.id AS users_id, foo FROM "
"orders JOIN users ON users.id = orders.user_id",
)
def test_backwards_join(self):
User, Address = self.classes.User, self.classes.Address
sess = fixture_session()
eq_(
sess.query(User)
.join(Address.user)
.filter(Address.email_address == "ed@wood.com")
.all(),
[User(id=8, name="ed")],
)
eq_(
sess.query(User, Address)
.join(Address.user)
.filter(Address.email_address == "ed@wood.com")
.all(),
[(User(id=8, name="ed"), Address(email_address="ed@wood.com"))],
)
assert_raises(
sa_exc.InvalidRequestError,
sess.query(User).join(Address, Address.user)._compile_context,
)
adalias = aliased(Address)
assert_raises(
sa_exc.InvalidRequestError,
sess.query(User).join(adalias, Address.user)._compile_context,
)
def test_multiple_with_aliases(self):
Order, User = self.classes.Order, self.classes.User
sess = fixture_session()
ualias = aliased(User)
oalias1 = aliased(Order)
oalias2 = aliased(Order)
self.assert_compile(
sess.query(ualias)
.join(oalias1, ualias.orders)
.join(oalias2, ualias.orders)
.filter(or_(oalias1.user_id == 9, oalias2.user_id == 7)),
"SELECT users_1.id AS users_1_id, users_1.name AS users_1_name "
"FROM users AS users_1 "
"JOIN orders AS orders_1 ON users_1.id = orders_1.user_id "
"JOIN orders AS orders_2 ON "
"users_1.id = orders_2.user_id "
"WHERE orders_1.user_id = :user_id_1 "
"OR orders_2.user_id = :user_id_2",
use_default_dialect=True,
)
def test_select_from_orm_joins(self):
User, Order = self.classes.User, self.classes.Order
sess = fixture_session()
ualias = aliased(User)
oalias1 = aliased(Order)
oalias2 = aliased(Order)
self.assert_compile(
join(User, oalias2, User.id == oalias2.user_id),
"users JOIN orders AS orders_1 ON users.id = orders_1.user_id",
use_default_dialect=True,
)
self.assert_compile(
join(User, oalias2, User.id == oalias2.user_id, full=True),
"users FULL OUTER JOIN orders AS orders_1 "
"ON users.id = orders_1.user_id",
use_default_dialect=True,
)
self.assert_compile(
join(User, oalias2, User.id == oalias2.user_id, isouter=True),
"users LEFT OUTER JOIN orders AS orders_1 "
"ON users.id = orders_1.user_id",
use_default_dialect=True,
)
self.assert_compile(
join(
User,
oalias2,
User.id == oalias2.user_id,
isouter=True,
full=True,
),
"users FULL OUTER JOIN orders AS orders_1 "
"ON users.id = orders_1.user_id",
use_default_dialect=True,
)
self.assert_compile(
join(User, oalias1).join(oalias2),
"users JOIN orders AS orders_1 ON users.id = orders_1.user_id "
"JOIN orders AS orders_2 ON users.id = orders_2.user_id",
use_default_dialect=True,
)
self.assert_compile(
join(User, oalias1).join(oalias2, isouter=True),
"users JOIN orders AS orders_1 ON users.id = orders_1.user_id "
"LEFT OUTER JOIN orders AS orders_2 "
"ON users.id = orders_2.user_id",
use_default_dialect=True,
)
self.assert_compile(
join(User, oalias1).join(oalias2, full=True),
"users JOIN orders AS orders_1 ON users.id = orders_1.user_id "
"FULL OUTER JOIN orders AS orders_2 "
"ON users.id = orders_2.user_id",
use_default_dialect=True,
)
self.assert_compile(
join(User, oalias1).join(oalias2, full=True, isouter=True),
"users JOIN orders AS orders_1 ON users.id = orders_1.user_id "
"FULL OUTER JOIN orders AS orders_2 "
"ON users.id = orders_2.user_id",
use_default_dialect=True,
)
self.assert_compile(
join(ualias, oalias1, ualias.orders),
"users AS users_1 JOIN orders AS orders_1 "
"ON users_1.id = orders_1.user_id",
use_default_dialect=True,
)
self.assert_compile(
sess.query(ualias).select_from(
join(ualias, oalias1, ualias.orders)
),
"SELECT users_1.id AS users_1_id, users_1.name AS users_1_name "
"FROM users AS users_1 "
"JOIN orders AS orders_1 ON users_1.id = orders_1.user_id",
use_default_dialect=True,
)
self.assert_compile(
sess.query(User, ualias).select_from(
join(ualias, oalias1, ualias.orders)
),
"SELECT users.id AS users_id, users.name AS users_name, "
"users_1.id AS users_1_id, "
"users_1.name AS users_1_name FROM users AS users_1 "
"JOIN orders AS orders_1 ON users_1.id = orders_1.user_id, users",
use_default_dialect=True,
)
if False:
self.assert_compile(
sess.query(User, ualias)
.join(oalias1, ualias.orders)
.join(oalias2, User.id == oalias2.user_id)
.filter(or_(oalias1.user_id == 9, oalias2.user_id == 7)),
"SELECT users.id AS users_id, users.name AS users_name, "
"users_1.id AS users_1_id, users_1.name AS "
"users_1_name FROM users JOIN orders AS orders_2 "
"ON users.id = orders_2.user_id, "
"users AS users_1 JOIN orders AS orders_1 "
"ON users_1.id = orders_1.user_id "
"WHERE orders_1.user_id = :user_id_1 "
"OR orders_2.user_id = :user_id_2",
use_default_dialect=True,
)
self.assert_compile(
sess.query(User, ualias)
.select_from(
join(ualias, oalias1, ualias.orders),
join(User, oalias2, User.id == oalias2.user_id),
)
.filter(or_(oalias1.user_id == 9, oalias2.user_id == 7)),
"SELECT users.id AS users_id, users.name AS users_name, "
"users_1.id AS users_1_id, users_1.name AS "
"users_1_name FROM users AS users_1 JOIN orders AS orders_1 "
"ON users_1.id = orders_1.user_id, "
"users JOIN orders AS orders_2 ON users.id = orders_2.user_id "
"WHERE orders_1.user_id = :user_id_1 "
"OR orders_2.user_id = :user_id_2",
use_default_dialect=True,
)
def test_overlapping_backwards_joins(self):
User, Order = self.classes.User, self.classes.Order
sess = fixture_session()
oalias1 = aliased(Order)
oalias2 = aliased(Order)
self.assert_compile(
sess.query(User).join(oalias1.user).join(oalias2.user),
"SELECT users.id AS users_id, users.name AS users_name "
"FROM orders AS orders_1 "
"JOIN users ON users.id = orders_1.user_id, orders AS orders_2 "
"JOIN users ON users.id = orders_2.user_id",
use_default_dialect=True,
)
def test_replace_multiple_from_clause(self):
"""test adding joins onto multiple FROM clauses"""
User, Order, Address = (
self.classes.User,
self.classes.Order,
self.classes.Address,
)
sess = fixture_session()
self.assert_compile(
sess.query(Address, User)
.join(Address.dingaling)
.join(User.orders)
.join(Order.items),
"SELECT addresses.id AS addresses_id, "
"addresses.user_id AS addresses_user_id, "
"addresses.email_address AS addresses_email_address, "
"users.id AS users_id, "
"users.name AS users_name FROM addresses JOIN dingalings "
"ON addresses.id = dingalings.address_id, "
"users JOIN orders ON users.id = orders.user_id "
"JOIN order_items AS order_items_1 "
"ON orders.id = order_items_1.order_id JOIN items "
"ON items.id = order_items_1.item_id",
use_default_dialect=True,
)
def test_invalid_join_entity_from_single_from_clause(self):
Address, Item = (self.classes.Address, self.classes.Item)
sess = fixture_session()
q = sess.query(Address).select_from(Address)
assert_raises_message(
sa.exc.InvalidRequestError,
"Don't know how to join to .*Item.*. "
r"Please use the .select_from\(\) "
"method to establish an explicit left side, as well as",
q.join(Item)._compile_context,
)
def test_invalid_join_entity_from_no_from_clause(self):
Address, Item = (self.classes.Address, self.classes.Item)
sess = fixture_session()
q = sess.query(Address)
assert_raises_message(
sa.exc.InvalidRequestError,
"Don't know how to join to .*Item.*. "
r"Please use the .select_from\(\) "
"method to establish an explicit left side, as well as",
q.join(Item)._compile_context,
)
def test_invalid_join_entity_from_multiple_from_clause(self):
"""test adding joins onto multiple FROM clauses where
we still need to say there's nothing to JOIN from"""
User, Address, Item = (
self.classes.User,
self.classes.Address,
self.classes.Item,
)
sess = fixture_session()
q = sess.query(Address, User).join(Address.dingaling).join(User.orders)
assert_raises_message(
sa.exc.InvalidRequestError,
"Don't know how to join to .*Item.*. "
r"Please use the .select_from\(\) "
"method to establish an explicit left side, as well as",
q.join(Item)._compile_context,
)
def test_join_explicit_left_multiple_from_clause(self):
"""test adding joins onto multiple FROM clauses where
it is ambiguous which FROM should be used when an
ON clause is given"""
User = self.classes.User
sess = fixture_session()
u1 = aliased(User)
q = sess.query(User, u1).select_from(User, u1).join(User.addresses)
self.assert_compile(
q,
"SELECT users.id AS users_id, users.name AS users_name, "
"users_1.id AS users_1_id, users_1.name AS users_1_name "
"FROM "
"users JOIN addresses ON users.id = addresses.user_id, "
"users AS users_1",
)
q = sess.query(User, u1).select_from(User, u1).join(u1.addresses)
self.assert_compile(
q,
"SELECT users.id AS users_id, users.name AS users_name, "
"users_1.id AS users_1_id, users_1.name AS users_1_name "
"FROM users, "
"users AS users_1 JOIN addresses "
"ON users_1.id = addresses.user_id",
)
def test_join_explicit_left_multiple_adapted(self):
"""test adding joins onto multiple FROM clauses where
it is ambiguous which FROM should be used when an
ON clause is given"""
User = self.classes.User
sess = fixture_session()
u1 = aliased(User)
u2 = aliased(User)
assert_raises_message(
sa_exc.InvalidRequestError,
"Can't identify which entity in which to assign the "
"left side of this join.",
sess.query(u1, u2)
.select_from(u1, u2)
.join(User.addresses)
._compile_context,
)
self.assert_compile(
sess.query(u1, u2).select_from(u1, u2).join(u2.addresses),
"SELECT users_1.id AS users_1_id, users_1.name AS users_1_name, "
"users_2.id AS users_2_id, users_2.name AS users_2_name "
"FROM users AS users_1, "
"users AS users_2 JOIN addresses "
"ON users_2.id = addresses.user_id",
)
def test_join_entity_from_multiple_from_clause(self):
"""test adding joins onto multiple FROM clauses where
it is ambiguous which FROM should be used"""
User, Order, Address, Dingaling = (
self.classes.User,
self.classes.Order,
self.classes.Address,
self.classes.Dingaling,
)
sess = fixture_session()
q = sess.query(Address, User).join(Address.dingaling).join(User.orders)
a1 = aliased(Address)
assert_raises_message(
sa.exc.InvalidRequestError,
"Can't determine which FROM clause to join from, there are "
"multiple FROMS which can join to this entity. "
r"Please use the .select_from\(\) "
"method to establish an explicit left side, as well as",
q.join(a1)._compile_context,
)
self.assert_compile(
q.join(a1, Order.address_id == a1.id),
"SELECT addresses.id AS addresses_id, "
"addresses.user_id AS addresses_user_id, "
"addresses.email_address AS addresses_email_address, "
"users.id AS users_id, users.name AS users_name "
"FROM addresses JOIN dingalings "
"ON addresses.id = dingalings.address_id, "
"users JOIN orders "
"ON users.id = orders.user_id "
"JOIN addresses AS addresses_1 "
"ON orders.address_id = addresses_1.id",
)
self.assert_compile(
q.join(a1, Dingaling.address_id == a1.id),
"SELECT addresses.id AS addresses_id, "
"addresses.user_id AS addresses_user_id, "
"addresses.email_address AS addresses_email_address, "
"users.id AS users_id, users.name AS users_name "
"FROM addresses JOIN dingalings "
"ON addresses.id = dingalings.address_id "
"JOIN addresses AS addresses_1 "
"ON dingalings.address_id = addresses_1.id, "
"users JOIN orders ON users.id = orders.user_id",
)
def test_join_entity_from_multiple_entities(self):
"""test adding joins onto multiple FROM clauses where
it is ambiguous which FROM should be used"""
Order, Address, Dingaling = (
self.classes.Order,
self.classes.Address,
self.classes.Dingaling,
)
sess = fixture_session()
q = sess.query(Order, Dingaling)
a1 = aliased(Address)
assert_raises_message(
sa.exc.InvalidRequestError,
"Can't determine which FROM clause to join from, there are "
"multiple FROMS which can join to this entity. "
r"Please use the .select_from\(\) "
"method to establish an explicit left side, as well as",
q.join(a1)._compile_context,
)
self.assert_compile(
q.join(a1, Order.address_id == a1.id),
"SELECT orders.id AS orders_id, orders.user_id AS orders_user_id, "
"orders.address_id AS orders_address_id, "
"orders.description AS orders_description, "
"orders.isopen AS orders_isopen, dingalings.id AS dingalings_id, "
"dingalings.address_id AS dingalings_address_id, "
"dingalings.data AS dingalings_data "
"FROM orders "
"JOIN addresses AS addresses_1 "
"ON orders.address_id = addresses_1.id, dingalings",
)
self.assert_compile(
q.join(a1, Dingaling.address_id == a1.id),
"SELECT orders.id AS orders_id, orders.user_id AS orders_user_id, "
"orders.address_id AS orders_address_id, "
"orders.description AS orders_description, "
"orders.isopen AS orders_isopen, dingalings.id AS dingalings_id, "
"dingalings.address_id AS dingalings_address_id, "
"dingalings.data AS dingalings_data "
"FROM dingalings JOIN addresses AS addresses_1 "
"ON dingalings.address_id = addresses_1.id, orders",
)
def test_clause_present_in_froms_twice_w_onclause(self):
Order, Address, User = (
self.classes.Order,
self.classes.Address,
self.classes.User,
)
sess = fixture_session()
a1 = aliased(Address)
q = sess.query(Order).select_from(Order, a1, User)
assert_raises_message(
sa.exc.InvalidRequestError,
"Can't determine which FROM clause to join from, there are "
"multiple FROMS which can join to this entity. "
r"Please use the .select_from\(\) "
"method to establish an explicit left side, as well as",
q.outerjoin(a1)._compile_context,
)
q = sess.query(Order).select_from(Order, a1, User)
q = q.outerjoin(a1, a1.id == Order.address_id)
q = q.outerjoin(User, a1.user_id == User.id)
self.assert_compile(
q,
"SELECT orders.id AS orders_id, orders.user_id AS orders_user_id, "
"orders.address_id AS orders_address_id, "
"orders.description AS orders_description, "
"orders.isopen AS orders_isopen "
"FROM orders "
"LEFT OUTER JOIN addresses AS addresses_1 "
"ON addresses_1.id = orders.address_id "
"LEFT OUTER JOIN users ON addresses_1.user_id = users.id",
)
def test_clause_present_in_froms_twice_wo_onclause(self):
Address, Dingaling, User = (
self.classes.Address,
self.classes.Dingaling,
self.classes.User,
)
sess = fixture_session()
a1 = aliased(Address)
q = sess.query(User).select_from(Dingaling, a1, User)
q = q.outerjoin(a1, User.id == a1.user_id)
q = q.outerjoin(Dingaling)
self.assert_compile(
q,
"SELECT users.id AS users_id, users.name AS users_name "
"FROM users LEFT OUTER JOIN addresses AS addresses_1 "
"ON users.id = addresses_1.user_id "
"LEFT OUTER JOIN dingalings "
"ON addresses_1.id = dingalings.address_id",
)
def test_pure_expression(self):
addresses, users = self.tables.addresses, self.tables.users
sess = fixture_session()
self.assert_compile(
sess.query(users).join(addresses),
"SELECT users.id AS users_id, users.name AS users_name "
"FROM users JOIN addresses ON users.id = addresses.user_id",
)
def test_no_onclause(self):
Item, User, Order = (
self.classes.Item,
self.classes.User,
self.classes.Order,
)
sess = fixture_session()
eq_(
sess.query(User)
.select_from(join(User, Order).join(Item, Order.items))
.filter(Item.description == "item 4")
.all(),
[User(name="jack")],
)
eq_(
sess.query(User.name)
.select_from(join(User, Order).join(Item, Order.items))
.filter(Item.description == "item 4")
.all(),
[("jack",)],
)
eq_(
sess.query(User)
.join(Order)
.join(Item, Order.items)
.filter(Item.description == "item 4")
.all(),
[User(name="jack")],
)
def test_clause_onclause(self):
Item, Order, order_items, User = (
self.classes.Item,
self.classes.Order,
self.tables.order_items,
self.classes.User,
)
sess = fixture_session()
eq_(
sess.query(User)
.join(Order, User.id == Order.user_id)
.join(order_items, Order.id == order_items.c.order_id)
.join(Item, order_items.c.item_id == Item.id)
.filter(Item.description == "item 4")
.all(),
[User(name="jack")],
)
eq_(
sess.query(User.name)
.join(Order, User.id == Order.user_id)
.join(order_items, Order.id == order_items.c.order_id)
.join(Item, order_items.c.item_id == Item.id)
.filter(Item.description == "item 4")
.all(),
[("jack",)],
)
ualias = aliased(User)
eq_(
sess.query(ualias.name)
.join(Order, ualias.id == Order.user_id)
.join(order_items, Order.id == order_items.c.order_id)
.join(Item, order_items.c.item_id == Item.id)
.filter(Item.description == "item 4")
.all(),
[("jack",)],
)
subq = sess.query(User).order_by(User.id).offset(2).subquery()
ua = aliased(User, subq)
eq_(
sess.query(ua).join(Order, ua.id == Order.user_id).all(),
[User(name="fred")],
)
def test_str_not_accepted_orm_join(self):
User, Address = self.classes.User, self.classes.Address
with expect_raises_message(
sa.exc.ArgumentError,
"ON clause, typically a SQL expression or ORM "
"relationship attribute expected, got 'addresses'.",
):
outerjoin(User, Address, "addresses")
def test_aliased_classes(self):
User, Address = self.classes.User, self.classes.Address
sess = fixture_session()
(user7, user8, user9, user10) = sess.query(User).all()
(address1, address2, address3, address4, address5) = sess.query(
Address
).all()
expected = [
(user7, address1),
(user8, address2),
(user8, address3),
(user8, address4),
(user9, address5),
(user10, None),
]
q = sess.query(User)
AdAlias = aliased(Address)
q = q.add_entity(AdAlias).select_from(outerjoin(User, AdAlias))
result = q.order_by(User.id, AdAlias.id).all()
eq_(result, expected)
sess.expunge_all()
q = sess.query(User).add_entity(AdAlias)
result = (
q.select_from(outerjoin(User, AdAlias))
.filter(AdAlias.email_address == "ed@bettyboop.com")
.all()
)
eq_(result, [(user8, address3)])
result = (
q.select_from(outerjoin(User, AdAlias, User.addresses))
.filter(AdAlias.email_address == "ed@bettyboop.com")
.all()
)
eq_(result, [(user8, address3)])
result = (
q.select_from(outerjoin(User, AdAlias, User.id == AdAlias.user_id))
.filter(AdAlias.email_address == "ed@bettyboop.com")
.all()
)
eq_(result, [(user8, address3)])
q = sess.query(User, AdAlias)
result = (
q.join(AdAlias.user)
.filter(User.name == "ed")
.order_by(User.id, AdAlias.id)
)
eq_(
result.all(),
[(user8, address2), (user8, address3), (user8, address4)],
)
q = (
sess.query(User, AdAlias)
.select_from(join(AdAlias, User, AdAlias.user))
.filter(User.name == "ed")
)
eq_(
result.all(),
[(user8, address2), (user8, address3), (user8, address4)],
)
def test_expression_onclauses(self):
Order, User = self.classes.Order, self.classes.User
sess = fixture_session()
subq = sess.query(User).subquery()
self.assert_compile(
sess.query(User).join(subq, User.name == subq.c.name),
"SELECT users.id AS users_id, users.name AS users_name "
"FROM users JOIN (SELECT users.id AS id, users.name "
"AS name FROM users) AS anon_1 ON users.name = anon_1.name",
use_default_dialect=True,
)
subq = sess.query(Order).subquery()
self.assert_compile(
sess.query(User).join(subq, User.id == subq.c.user_id),
"SELECT users.id AS users_id, users.name AS users_name FROM "
"users JOIN (SELECT orders.id AS id, orders.user_id AS user_id, "
"orders.address_id AS address_id, orders.description AS "
"description, orders.isopen AS isopen FROM orders) AS "
"anon_1 ON users.id = anon_1.user_id",
use_default_dialect=True,
)
self.assert_compile(
sess.query(User).join(Order, User.id == Order.user_id),
"SELECT users.id AS users_id, users.name AS users_name "
"FROM users JOIN orders ON users.id = orders.user_id",
use_default_dialect=True,
)
def test_aliased_classes_m2m(self):
Item, Order = self.classes.Item, self.classes.Order
sess = fixture_session()
(order1, order2, order3, order4, order5) = sess.query(Order).all()
(item1, item2, item3, item4, item5) = sess.query(Item).all()
expected = [
(order1, item1),
(order1, item2),
(order1, item3),
(order2, item1),
(order2, item2),
(order2, item3),
(order3, item3),
(order3, item4),
(order3, item5),
(order4, item1),
(order4, item5),
(order5, item5),
]
q = sess.query(Order)
q = (
q.add_entity(Item)
.select_from(join(Order, Item, Order.items))
.order_by(Order.id, Item.id)
)
result = q.all()
eq_(result, expected)
IAlias = aliased(Item)
q = (
sess.query(Order, IAlias)
.select_from(join(Order, IAlias, Order.items))
.filter(IAlias.description == "item 3")
)
result = q.all()
eq_(result, [(order1, item3), (order2, item3), (order3, item3)])
def test_joins_from_adapted_entities(self):
User = self.classes.User
session = fixture_session()
first = session.query(User)
second = session.query(User)
unioned = first.union(second)
subquery = session.query(User.id).subquery()
join = subquery, subquery.c.id == User.id
joined = unioned.outerjoin(*join)
self.assert_compile(
joined,
"SELECT anon_1.users_id AS "
"anon_1_users_id, anon_1.users_name AS "
"anon_1_users_name FROM (SELECT users.id "
"AS users_id, users.name AS users_name "
"FROM users UNION SELECT users.id AS "
"users_id, users.name AS users_name FROM "
"users) AS anon_1 LEFT OUTER JOIN (SELECT "
"users.id AS id FROM users) AS anon_2 ON "
"anon_2.id = anon_1.users_id",
use_default_dialect=True,
)
first = session.query(User.id)
second = session.query(User.id)
unioned = first.union(second)
subquery = session.query(User.id).subquery()
join = subquery, subquery.c.id == User.id
joined = unioned.outerjoin(*join)
self.assert_compile(
joined,
"SELECT anon_1.users_id AS anon_1_users_id "
"FROM (SELECT users.id AS users_id FROM "
"users UNION SELECT users.id AS users_id "
"FROM users) AS anon_1 LEFT OUTER JOIN "
"(SELECT users.id AS id FROM users) AS "
"anon_2 ON anon_2.id = anon_1.users_id",
use_default_dialect=True,
)
def test_joins_from_adapted_entities_isouter(self):
User = self.classes.User
session = fixture_session()
first = session.query(User)
second = session.query(User)
unioned = first.union(second)
subquery = session.query(User.id).subquery()
join = subquery, subquery.c.id == User.id
joined = unioned.join(*join, isouter=True)
self.assert_compile(
joined,
"SELECT anon_1.users_id AS "
"anon_1_users_id, anon_1.users_name AS "
"anon_1_users_name FROM (SELECT users.id "
"AS users_id, users.name AS users_name "
"FROM users UNION SELECT users.id AS "
"users_id, users.name AS users_name FROM "
"users) AS anon_1 LEFT OUTER JOIN (SELECT "
"users.id AS id FROM users) AS anon_2 ON "
"anon_2.id = anon_1.users_id",
use_default_dialect=True,
)
first = session.query(User.id)
second = session.query(User.id)
unioned = first.union(second)
subquery = session.query(User.id).subquery()
join = subquery, subquery.c.id == User.id
joined = unioned.join(*join, isouter=True)
self.assert_compile(
joined,
"SELECT anon_1.users_id AS anon_1_users_id "
"FROM (SELECT users.id AS users_id FROM "
"users UNION SELECT users.id AS users_id "
"FROM users) AS anon_1 LEFT OUTER JOIN "
"(SELECT users.id AS id FROM users) AS "
"anon_2 ON anon_2.id = anon_1.users_id",
use_default_dialect=True,
)
def test_overlap_with_aliases(self):
orders, User, users = (
self.tables.orders,
self.classes.User,
self.tables.users,
)
Order = self.classes.Order
oalias = orders.alias("oalias")
result = (
fixture_session()
.query(User)
.select_from(users.join(oalias))
.filter(
oalias.c.description.in_(["order 1", "order 2", "order 3"])
)
.join(User.orders)
.join(Order.items)
.order_by(User.id)
.all()
)
assert [User(id=7, name="jack"), User(id=9, name="fred")] == result
result = (
fixture_session()
.query(User)
.select_from(users.join(oalias))
.filter(
oalias.c.description.in_(["order 1", "order 2", "order 3"])
)
.join(User.orders)
.join(Order.items)
.filter_by(id=4)
.all()
)
assert [User(id=7, name="jack")] == result
def test_aliased_order_by(self):
User = self.classes.User
sess = fixture_session()
ualias = aliased(User)
eq_(
sess.query(User, ualias)
.filter(User.id > ualias.id)
.order_by(desc(ualias.id), User.name)
.all(),
[
(User(id=10, name="chuck"), User(id=9, name="fred")),
(User(id=10, name="chuck"), User(id=8, name="ed")),
(User(id=9, name="fred"), User(id=8, name="ed")),
(User(id=10, name="chuck"), User(id=7, name="jack")),
(User(id=8, name="ed"), User(id=7, name="jack")),
(User(id=9, name="fred"), User(id=7, name="jack")),
],
)
def test_plain_table(self):
addresses, User = self.tables.addresses, self.classes.User
sess = fixture_session()
eq_(
sess.query(User.name)
.join(addresses, User.id == addresses.c.user_id)
.order_by(User.id)
.all(),
[("jack",), ("ed",), ("ed",), ("ed",), ("fred",)],
)
def test_no_joinpoint_expr(self):
User, users = self.classes.User, self.tables.users
sess = fixture_session()
assert_raises_message(
sa_exc.InvalidRequestError,
"Don't know how to join to .*User.*. "
r"Please use the .select_from\(\) "
"method to establish an explicit left side, as well as",
sess.query(users.c.id).join(User)._compile_context,
)
assert_raises_message(
sa_exc.InvalidRequestError,
"Don't know how to join to .*User.* "
r"Please use the .select_from\(\) "
"method to establish an explicit left side, as well as",
sess.query(users.c.id)
.select_from(users)
.join(User)
._compile_context,
)
def test_on_clause_no_right_side_one(self):
User = self.classes.User
Address = self.classes.Address
sess = fixture_session()
with expect_raises_message(
sa_exc.ArgumentError,
"Join target, typically a FROM expression, or ORM relationship "
"attribute expected, got",
):
sess.query(User).join(User.id == Address.user_id)
def test_on_clause_no_right_side_one_future(self):
User = self.classes.User
Address = self.classes.Address
assert_raises_message(
sa_exc.ArgumentError,
"Join target, typically a FROM expression, "
"or ORM relationship attribute expected",
select(User).join,
User.id == Address.user_id,
)
def test_no_legacy_multi_join_two_element(self):
User = self.classes.User
Order = self.classes.Order
sess = fixture_session()
with expect_raises_message(
sa_exc.InvalidRequestError,
"No 'on clause' argument may be passed when joining to a "
"relationship path as a target",
):
sess.query(User).join(User.orders, Order.items)._compile_context()
def test_no_modern_multi_join_two_element(self):
User = self.classes.User
Order = self.classes.Order
sess = fixture_session()
with expect_raises_message(
sa_exc.InvalidRequestError,
"No 'on clause' argument may be passed when joining to a "
"relationship path as a target",
):
sess.execute(select(User).join(User.orders, Order.items))
def test_kw_only_blocks_legacy_multi_join(self):
User = self.classes.User
Order = self.classes.Order
Item = self.classes.Item
sess = fixture_session()
with expect_raises_message(
TypeError,
r".*join\(\) takes from 2 to 3 positional arguments but "
"4 were given",
):
sess.query(User).join(User.orders, Order.items, Item.keywords)
def test_on_clause_no_right_side_two(self):
User = self.classes.User
Address = self.classes.Address
sess = fixture_session()
assert_raises_message(
sa_exc.ArgumentError,
"Join target Address.user_id does not refer to a mapped entity",
sess.query(User).join(Address.user_id)._compile_context,
)
def test_on_clause_no_right_side_two_future(self):
User = self.classes.User
Address = self.classes.Address
stmt = select(User).join(Address.user_id)
assert_raises_message(
sa_exc.ArgumentError,
"Join target Address.user_id does not refer to a mapped entity",
stmt.compile,
)
def test_no_strings_for_single_onclause_legacy_query(self):
User = self.classes.User
sess = fixture_session()
with expect_raises_message(
sa_exc.ArgumentError,
"Join target, typically a FROM expression, or ORM relationship "
"attribute expected, got 'addresses'",
):
sess.query(User).join("addresses")
def test_no_strings_for_single_onclause_newstyle(self):
User = self.classes.User
with expect_raises_message(
sa_exc.ArgumentError,
"Join target, typically a FROM expression, or ORM relationship "
"attribute expected, got 'addresses'",
):
select(User).join("addresses")
def test_no_strings_for_dual_onclause_legacy_query(self):
User = self.classes.User
Address = self.classes.Address
sess = fixture_session()
with expect_raises_message(
sa_exc.ArgumentError,
"ON clause, typically a SQL expression or ORM relationship "
"attribute expected, got 'addresses'",
):
sess.query(User).join(Address, "addresses")
def test_no_strings_for_dual_onclause_newstyle(self):
User = self.classes.User
Address = self.classes.Address
with expect_raises_message(
sa_exc.ArgumentError,
"ON clause, typically a SQL expression or ORM relationship "
"attribute expected, got 'addresses'.",
):
select(User).join(Address, "addresses")
def test_select_from(self):
"""Test that the left edge of the join can be set reliably with
select_from()."""
Item, Order, User = (
self.classes.Item,
self.classes.Order,
self.classes.User,
)
sess = fixture_session()
self.assert_compile(
sess.query(Item.id)
.select_from(User)
.join(User.orders)
.join(Order.items),
"SELECT items.id AS items_id FROM users JOIN orders ON "
"users.id = orders.user_id JOIN order_items AS order_items_1 "
"ON orders.id = order_items_1.order_id JOIN items ON items.id = "
"order_items_1.item_id",
use_default_dialect=True,
)
self.assert_compile(
sess.query(Item.id)
.select_from(User)
.join(Item, User.id == Item.id),
"SELECT items.id AS items_id FROM users JOIN items "
"ON users.id = items.id",
use_default_dialect=True,
)
class JoinFromSelectableTest(fixtures.MappedTest, AssertsCompiledSQL):
__dialect__ = "default"
run_setup_mappers = "once"
@classmethod
def define_tables(cls, metadata):
Table("table1", metadata, Column("id", Integer, primary_key=True))
Table(
"table2",
metadata,
Column("id", Integer, primary_key=True),
Column("t1_id", Integer),
)
@classmethod
def setup_classes(cls):
class T1(cls.Comparable):
pass
class T2(cls.Comparable):
pass
@classmethod
def setup_mappers(cls):
table1, table2 = cls.tables.table1, cls.tables.table2
T1, T2 = cls.classes("T1", "T2")
cls.mapper_registry.map_imperatively(T1, table1)
cls.mapper_registry.map_imperatively(T2, table2)
def test_select_mapped_to_mapped_explicit_left(self):
T1, T2 = self.classes.T1, self.classes.T2
sess = fixture_session()
subq = (
sess.query(T2.t1_id, func.count(T2.id).label("count"))
.group_by(T2.t1_id)
.subquery()
)
self.assert_compile(
sess.query(subq.c.count, T1.id)
.select_from(subq)
.join(T1, subq.c.t1_id == T1.id),
"SELECT anon_1.count AS anon_1_count, table1.id AS table1_id "
"FROM (SELECT table2.t1_id AS t1_id, "
"count(table2.id) AS count FROM table2 "
"GROUP BY table2.t1_id) AS anon_1 JOIN table1 "
"ON anon_1.t1_id = table1.id",
)
def test_select_mapped_to_mapped_implicit_left(self):
T1, T2 = self.classes.T1, self.classes.T2
sess = fixture_session()
subq = (
sess.query(T2.t1_id, func.count(T2.id).label("count"))
.group_by(T2.t1_id)
.subquery()
)
self.assert_compile(
sess.query(subq.c.count, T1.id).join(T1, subq.c.t1_id == T1.id),
"SELECT anon_1.count AS anon_1_count, table1.id AS table1_id "
"FROM (SELECT table2.t1_id AS t1_id, "
"count(table2.id) AS count FROM table2 "
"GROUP BY table2.t1_id) AS anon_1 JOIN table1 "
"ON anon_1.t1_id = table1.id",
)
def test_select_mapped_to_select_explicit_left(self):
T1, T2 = self.classes.T1, self.classes.T2
sess = fixture_session()
subq = (
sess.query(T2.t1_id, func.count(T2.id).label("count"))
.group_by(T2.t1_id)
.subquery()
)
self.assert_compile(
sess.query(subq.c.count, T1.id)
.select_from(T1)
.join(subq, subq.c.t1_id == T1.id),
"SELECT anon_1.count AS anon_1_count, table1.id AS table1_id "
"FROM table1 JOIN (SELECT table2.t1_id AS t1_id, "
"count(table2.id) AS count FROM table2 GROUP BY table2.t1_id) "
"AS anon_1 ON anon_1.t1_id = table1.id",
)
def test_select_mapped_to_select_implicit_left(self):
T1, T2 = self.classes.T1, self.classes.T2
sess = fixture_session()
subq = (
sess.query(T2.t1_id, func.count(T2.id).label("count"))
.group_by(T2.t1_id)
.subquery()
)
self.assert_compile(
sess.query(subq.c.count, T1.id).join(subq, subq.c.t1_id == T1.id),
"SELECT anon_1.count AS anon_1_count, table1.id AS table1_id "
"FROM table1 JOIN "
"(SELECT table2.t1_id AS t1_id, count(table2.id) AS count "
"FROM table2 GROUP BY table2.t1_id) "
"AS anon_1 ON anon_1.t1_id = table1.id",
)
self.assert_compile(
sess.query(subq.c.count, T1.id)
.select_from(T1)
.join(subq, subq.c.t1_id == T1.id),
"SELECT anon_1.count AS anon_1_count, table1.id AS table1_id "
"FROM table1 JOIN "
"(SELECT table2.t1_id AS t1_id, count(table2.id) AS count "
"FROM table2 GROUP BY table2.t1_id) "
"AS anon_1 ON anon_1.t1_id = table1.id",
)
def test_mapped_select_to_mapped_implicit_left(self):
T1, T2 = self.classes.T1, self.classes.T2
sess = fixture_session()
subq = (
sess.query(T2.t1_id, func.count(T2.id).label("count"))
.group_by(T2.t1_id)
.subquery()
)
self.assert_compile(
sess.query(T1.id, subq.c.count).join(T1, subq.c.t1_id == T1.id),
"SELECT table1.id AS table1_id, anon_1.count AS anon_1_count "
"FROM (SELECT table2.t1_id AS t1_id, count(table2.id) AS count "
"FROM table2 GROUP BY table2.t1_id) AS anon_1 "
"JOIN table1 ON anon_1.t1_id = table1.id",
)
self.assert_compile(
sess.query(T1.id, subq.c.count)
.select_from(subq)
.join(T1, subq.c.t1_id == T1.id),
"SELECT table1.id AS table1_id, anon_1.count AS anon_1_count "
"FROM (SELECT table2.t1_id AS t1_id, count(table2.id) AS count "
"FROM table2 GROUP BY table2.t1_id) AS anon_1 "
"JOIN table1 ON anon_1.t1_id = table1.id",
)
def test_mapped_select_to_mapped_explicit_left(self):
T1, T2 = self.classes.T1, self.classes.T2
sess = fixture_session()
subq = (
sess.query(T2.t1_id, func.count(T2.id).label("count"))
.group_by(T2.t1_id)
.subquery()
)
self.assert_compile(
sess.query(T1.id, subq.c.count)
.select_from(subq)
.join(T1, subq.c.t1_id == T1.id),
"SELECT table1.id AS table1_id, anon_1.count AS anon_1_count "
"FROM (SELECT table2.t1_id AS t1_id, count(table2.id) AS count "
"FROM table2 GROUP BY table2.t1_id) AS anon_1 JOIN table1 "
"ON anon_1.t1_id = table1.id",
)
def test_mapped_select_to_select_explicit_left(self):
T1, T2 = self.classes.T1, self.classes.T2
sess = fixture_session()
subq = (
sess.query(T2.t1_id, func.count(T2.id).label("count"))
.group_by(T2.t1_id)
.subquery()
)
self.assert_compile(
sess.query(T1.id, subq.c.count)
.select_from(T1)
.join(subq, subq.c.t1_id == T1.id),
"SELECT table1.id AS table1_id, anon_1.count AS anon_1_count "
"FROM table1 JOIN (SELECT table2.t1_id AS t1_id, "
"count(table2.id) AS count "
"FROM table2 GROUP BY table2.t1_id) AS anon_1 "
"ON anon_1.t1_id = table1.id",
)
def test_mapped_select_to_select_implicit_left(self):
T1, T2 = self.classes.T1, self.classes.T2
sess = fixture_session()
subq = (
sess.query(T2.t1_id, func.count(T2.id).label("count"))
.group_by(T2.t1_id)
.subquery()
)
self.assert_compile(
sess.query(T1.id, subq.c.count).join(subq, subq.c.t1_id == T1.id),
"SELECT table1.id AS table1_id, anon_1.count AS anon_1_count "
"FROM table1 JOIN (SELECT table2.t1_id AS t1_id, "
"count(table2.id) AS count "
"FROM table2 GROUP BY table2.t1_id) AS anon_1 "
"ON anon_1.t1_id = table1.id",
)
class SelfRefMixedTest(fixtures.MappedTest, AssertsCompiledSQL):
run_setup_mappers = "once"
__dialect__ = default.DefaultDialect()
@classmethod
def define_tables(cls, metadata):
Table(
"nodes",
metadata,
Column(
"id", Integer, primary_key=True, test_needs_autoincrement=True
),
Column("parent_id", Integer, ForeignKey("nodes.id")),
)
Table(
"sub_table",
metadata,
Column(
"id", Integer, primary_key=True, test_needs_autoincrement=True
),
Column("node_id", Integer, ForeignKey("nodes.id")),
)
Table(
"assoc_table",
metadata,
Column("left_id", Integer, ForeignKey("nodes.id")),
Column("right_id", Integer, ForeignKey("nodes.id")),
)
@classmethod
def setup_classes(cls):
class Node(cls.Comparable):
pass
class Sub(cls.Comparable):
pass
@classmethod
def setup_mappers(cls):
nodes, assoc_table, sub_table = (
cls.tables.nodes,
cls.tables.assoc_table,
cls.tables.sub_table,
)
Node, Sub = cls.classes("Node", "Sub")
cls.mapper_registry.map_imperatively(
Node,
nodes,
properties={
"children": relationship(
Node,
lazy="select",
join_depth=3,
backref=backref("parent", remote_side=[nodes.c.id]),
),
"subs": relationship(Sub),
"assoc": relationship(
Node,
secondary=assoc_table,
primaryjoin=nodes.c.id == assoc_table.c.left_id,
secondaryjoin=nodes.c.id == assoc_table.c.right_id,
),
},
)
cls.mapper_registry.map_imperatively(Sub, sub_table)
def test_o2m_aliased_plus_o2m(self):
Node, Sub = self.classes.Node, self.classes.Sub
sess = fixture_session()
n1 = aliased(Node)
self.assert_compile(
sess.query(Node).join(n1, Node.children).join(Sub, n1.subs),
"SELECT nodes.id AS nodes_id, nodes.parent_id AS nodes_parent_id "
"FROM nodes JOIN nodes AS nodes_1 ON nodes.id = nodes_1.parent_id "
"JOIN sub_table ON nodes_1.id = sub_table.node_id",
)
self.assert_compile(
sess.query(Node).join(n1, Node.children).join(Sub, Node.subs),
"SELECT nodes.id AS nodes_id, nodes.parent_id AS nodes_parent_id "
"FROM nodes JOIN nodes AS nodes_1 ON nodes.id = nodes_1.parent_id "
"JOIN sub_table ON nodes.id = sub_table.node_id",
)
def test_m2m_aliased_plus_o2m(self):
Node, Sub = self.classes.Node, self.classes.Sub
sess = fixture_session()
n1 = aliased(Node)
self.assert_compile(
sess.query(Node).join(n1, Node.assoc).join(Sub, n1.subs),
"SELECT nodes.id AS nodes_id, nodes.parent_id AS nodes_parent_id "
"FROM nodes JOIN assoc_table AS assoc_table_1 ON nodes.id = "
"assoc_table_1.left_id JOIN nodes AS nodes_1 ON nodes_1.id = "
"assoc_table_1.right_id JOIN sub_table "
"ON nodes_1.id = sub_table.node_id",
)
self.assert_compile(
sess.query(Node).join(n1, Node.assoc).join(Sub, Node.subs),
"SELECT nodes.id AS nodes_id, nodes.parent_id AS nodes_parent_id "
"FROM nodes JOIN assoc_table AS assoc_table_1 ON nodes.id = "
"assoc_table_1.left_id JOIN nodes AS nodes_1 ON nodes_1.id = "
"assoc_table_1.right_id JOIN sub_table "
"ON nodes.id = sub_table.node_id",
)
class CreateJoinsTest(fixtures.MappedTest, AssertsCompiledSQL):
__dialect__ = "default"
def _inherits_fixture(self):
m = MetaData()
base = Table("base", m, Column("id", Integer, primary_key=True))
a = Table(
"a",
m,
Column("id", Integer, ForeignKey("base.id"), primary_key=True),
Column("b_id", Integer, ForeignKey("b.id")),
)
b = Table(
"b",
m,
Column("id", Integer, ForeignKey("base.id"), primary_key=True),
Column("c_id", Integer, ForeignKey("c.id")),
)
c = Table(
"c",
m,
Column("id", Integer, ForeignKey("base.id"), primary_key=True),
)
class Base:
pass
class A(Base):
pass
class B(Base):
pass
class C(Base):
pass
self.mapper_registry.map_imperatively(Base, base)
self.mapper_registry.map_imperatively(
A,
a,
inherits=Base,
properties={"b": relationship(B, primaryjoin=a.c.b_id == b.c.id)},
)
self.mapper_registry.map_imperatively(
B,
b,
inherits=Base,
properties={"c": relationship(C, primaryjoin=b.c.c_id == c.c.id)},
)
self.mapper_registry.map_imperatively(C, c, inherits=Base)
return A, B, C, Base
def test_double_level_aliased_exists(self):
A, B, C, Base = self._inherits_fixture()
s = fixture_session()
self.assert_compile(
s.query(A).filter(A.b.has(B.c.has(C.id == 5))),
"SELECT a.id AS a_id, base.id AS base_id, a.b_id AS a_b_id "
"FROM base JOIN a ON base.id = a.id WHERE "
"EXISTS (SELECT 1 FROM (SELECT base.id AS base_id, b.id AS "
"b_id, b.c_id AS b_c_id FROM base JOIN b ON base.id = b.id) "
"AS anon_1 WHERE a.b_id = anon_1.b_id AND (EXISTS "
"(SELECT 1 FROM (SELECT base.id AS base_id, c.id AS c_id "
"FROM base JOIN c ON base.id = c.id) AS anon_2 "
"WHERE anon_1.b_c_id = anon_2.c_id AND anon_2.c_id = :id_1"
")))",
)
class JoinToNonPolyAliasesTest(fixtures.MappedTest, AssertsCompiledSQL):
"""test joins to an aliased selectable and that we can refer to that
aliased selectable in filter criteria.
Basically testing that the aliasing Query applies to with_polymorphic
targets doesn't leak into non-polymorphic mappers.
"""
__dialect__ = "default"
run_create_tables = None
run_deletes = None
@classmethod
def define_tables(cls, metadata):
Table(
"parent",
metadata,
Column("id", Integer, primary_key=True),
Column("data", String(50)),
)
Table(
"child",
metadata,
Column("id", Integer, primary_key=True),
Column("parent_id", Integer, ForeignKey("parent.id")),
Column("data", String(50)),
)
@classmethod
def setup_mappers(cls):
parent, child = cls.tables.parent, cls.tables.child
class Parent(cls.Comparable):
pass
class Child(cls.Comparable):
pass
mp = cls.mapper_registry.map_imperatively(Parent, parent)
cls.mapper_registry.map_imperatively(Child, child)
derived = select(child).alias()
npc = aliased(Child, derived)
cls.npc = npc
cls.derived = derived
mp.add_property("npc", relationship(npc))
def test_join_parent_child(self):
Parent = self.classes.Parent
sess = fixture_session()
self.assert_compile(
sess.query(Parent)
.join(Parent.npc)
.filter(self.derived.c.data == "x"),
"SELECT parent.id AS parent_id, parent.data AS parent_data "
"FROM parent JOIN (SELECT child.id AS id, "
"child.parent_id AS parent_id, "
"child.data AS data "
"FROM child) AS anon_1 ON parent.id = anon_1.parent_id "
"WHERE anon_1.data = :data_1",
)
def test_join_parent_child_select_from(self):
Parent = self.classes.Parent
npc = self.npc
sess = fixture_session()
self.assert_compile(
sess.query(npc)
.select_from(Parent)
.join(Parent.npc)
.filter(self.derived.c.data == "x"),
"SELECT anon_1.id AS anon_1_id, anon_1.parent_id "
"AS anon_1_parent_id, anon_1.data AS anon_1_data "
"FROM parent JOIN (SELECT child.id AS id, child.parent_id AS "
"parent_id, child.data AS data FROM child) AS anon_1 ON "
"parent.id = anon_1.parent_id WHERE anon_1.data = :data_1",
)
def test_join_select_parent_child(self):
Parent = self.classes.Parent
npc = self.npc
sess = fixture_session()
self.assert_compile(
sess.query(Parent, npc)
.join(Parent.npc)
.filter(self.derived.c.data == "x"),
"SELECT parent.id AS parent_id, parent.data AS parent_data, "
"anon_1.id AS anon_1_id, anon_1.parent_id AS anon_1_parent_id, "
"anon_1.data AS anon_1_data FROM parent JOIN "
"(SELECT child.id AS id, child.parent_id AS parent_id, "
"child.data AS data FROM child) AS anon_1 ON parent.id = "
"anon_1.parent_id WHERE anon_1.data = :data_1",
)
class SelfReferentialTest(fixtures.MappedTest, AssertsCompiledSQL):
run_setup_mappers = "once"
run_inserts = "once"
run_deletes = None
__dialect__ = "default"
@classmethod
def define_tables(cls, metadata):
Table(
"nodes",
metadata,
Column(
"id", Integer, primary_key=True, test_needs_autoincrement=True
),
Column("parent_id", Integer, ForeignKey("nodes.id")),
Column("data", String(30)),
)
@classmethod
def setup_classes(cls):
class Node(cls.Comparable):
def append(self, node):
self.children.append(node)
@classmethod
def setup_mappers(cls):
Node, nodes = cls.classes.Node, cls.tables.nodes
cls.mapper_registry.map_imperatively(
Node,
nodes,
properties={
"children": relationship(
Node,
lazy="select",
join_depth=3,
backref=backref("parent", remote_side=[nodes.c.id]),
)
},
)
@classmethod
def insert_data(cls, connection):
Node = cls.classes.Node
sess = Session(connection)
n1 = Node(data="n1")
n1.append(Node(data="n11"))
n1.append(Node(data="n12"))
n1.append(Node(data="n13"))
n1.children[1].append(Node(data="n121"))
n1.children[1].append(Node(data="n122"))
n1.children[1].append(Node(data="n123"))
sess.add(n1)
sess.flush()
sess.close()
def test_join_4_explicit_join(self):
Node = self.classes.Node
sess = fixture_session()
na = aliased(Node)
na2 = aliased(Node)
q = (
sess.query(Node)
.filter(Node.data == "n122")
.join(na, Node.parent)
.filter(na.data == "n12")
.join(na2, na.parent)
.filter(na2.data == "n1")
)
self.assert_compile(
q,
"SELECT nodes.id AS nodes_id, nodes.parent_id AS nodes_parent_id, "
"nodes.data AS nodes_data FROM nodes JOIN nodes AS nodes_1 "
"ON nodes_1.id = nodes.parent_id JOIN nodes AS nodes_2 "
"ON nodes_2.id = nodes_1.parent_id WHERE nodes.data = :data_1 "
"AND nodes_1.data = :data_2 AND nodes_2.data = :data_3",
checkparams={"data_1": "n122", "data_2": "n12", "data_3": "n1"},
)
node = q.first()
eq_(node.data, "n122")
def test_from_self_inside_excludes_outside(self):
"""test the propagation of aliased() from inside to outside
on a from_self()..
"""
Node = self.classes.Node
sess = fixture_session()
n1 = aliased(Node)
subq = (
sess.query(Node)
.filter(Node.data == "n122")
.set_label_style(LABEL_STYLE_TABLENAME_PLUS_COL)
.subquery()
)
na = aliased(Node, subq)
self.assert_compile(
sess.query(n1, na.id),
"SELECT nodes_1.id AS nodes_1_id, "
"nodes_1.parent_id AS nodes_1_parent_id, "
"nodes_1.data AS nodes_1_data, anon_1.nodes_id AS anon_1_nodes_id "
"FROM nodes AS nodes_1, (SELECT nodes.id AS nodes_id, "
"nodes.parent_id AS nodes_parent_id, "
"nodes.data AS nodes_data FROM "
"nodes WHERE nodes.data = :data_1) AS anon_1",
use_default_dialect=True,
)
parent = aliased(Node)
grandparent = aliased(Node)
subq = (
sess.query(Node, parent, grandparent)
.join(parent, Node.parent)
.join(grandparent, parent.parent)
.filter(Node.data == "n122")
.filter(parent.data == "n12")
.filter(grandparent.data == "n1")
.set_label_style(LABEL_STYLE_TABLENAME_PLUS_COL)
.subquery()
)
na = aliased(Node, subq)
pa = aliased(parent, subq)
ga = aliased(grandparent, subq)
q = sess.query(na, pa, ga).limit(1)
self.assert_compile(
q,
"SELECT anon_1.nodes_id AS anon_1_nodes_id, "
"anon_1.nodes_parent_id AS anon_1_nodes_parent_id, "
"anon_1.nodes_data AS anon_1_nodes_data, "
"anon_1.nodes_1_id AS anon_1_nodes_1_id, "
"anon_1.nodes_1_parent_id AS anon_1_nodes_1_parent_id, "
"anon_1.nodes_1_data AS anon_1_nodes_1_data, "
"anon_1.nodes_2_id AS anon_1_nodes_2_id, "
"anon_1.nodes_2_parent_id AS anon_1_nodes_2_parent_id, "
"anon_1.nodes_2_data AS anon_1_nodes_2_data "
"FROM (SELECT nodes.id AS nodes_id, nodes.parent_id "
"AS nodes_parent_id, nodes.data AS nodes_data, "
"nodes_1.id AS nodes_1_id, "
"nodes_1.parent_id AS nodes_1_parent_id, "
"nodes_1.data AS nodes_1_data, nodes_2.id AS nodes_2_id, "
"nodes_2.parent_id AS nodes_2_parent_id, nodes_2.data AS "
"nodes_2_data FROM nodes JOIN nodes AS nodes_1 ON "
"nodes_1.id = nodes.parent_id JOIN nodes AS nodes_2 "
"ON nodes_2.id = nodes_1.parent_id "
"WHERE nodes.data = :data_1 AND nodes_1.data = :data_2 AND "
"nodes_2.data = :data_3) AS anon_1 LIMIT :param_1",
{"param_1": 1},
use_default_dialect=True,
)
def test_join_to_self_no_aliases_raises(self):
Node = self.classes.Node
s = fixture_session()
assert_raises_message(
sa.exc.InvalidRequestError,
r"Can't construct a join from Mapper\[Node\(nodes\)\] to "
r"Mapper\[Node\(nodes\)\], they are the same entity",
s.query(Node).join(Node.children)._compile_context,
)
def test_explicit_join_2(self):
Node = self.classes.Node
n1 = aliased(Node)
n2 = aliased(Node)
self.assert_compile(
join(Node, n1, Node.children).join(n2, n1.children),
"nodes JOIN nodes AS nodes_1 ON nodes.id = nodes_1.parent_id "
"JOIN nodes AS nodes_2 ON nodes_1.id = nodes_2.parent_id",
use_default_dialect=True,
)
def test_explicit_join_3(self):
Node = self.classes.Node
n1 = aliased(Node)
n2 = aliased(Node)
self.assert_compile(
join(Node, n1, Node.children).join(n2, Node.children),
"nodes JOIN nodes AS nodes_1 ON nodes.id = nodes_1.parent_id "
"JOIN nodes AS nodes_2 ON nodes.id = nodes_2.parent_id",
use_default_dialect=True,
)
def test_explicit_join_4(self):
Node = self.classes.Node
sess = fixture_session()
n1 = aliased(Node)
n2 = aliased(Node)
self.assert_compile(
sess.query(Node).join(n1, Node.children).join(n2, n1.children),
"SELECT nodes.id AS nodes_id, nodes.parent_id AS nodes_parent_id, "
"nodes.data AS nodes_data FROM nodes JOIN nodes AS nodes_1 "
"ON nodes.id = nodes_1.parent_id "
"JOIN nodes AS nodes_2 ON nodes_1.id = nodes_2.parent_id",
use_default_dialect=True,
)
def test_explicit_join_5(self):
Node = self.classes.Node
sess = fixture_session()
n1 = aliased(Node)
n2 = aliased(Node)
self.assert_compile(
sess.query(Node).join(n1, Node.children).join(n2, Node.children),
"SELECT nodes.id AS nodes_id, nodes.parent_id AS nodes_parent_id, "
"nodes.data AS nodes_data FROM nodes JOIN nodes AS nodes_1 "
"ON nodes.id = nodes_1.parent_id "
"JOIN nodes AS nodes_2 ON nodes.id = nodes_2.parent_id",
use_default_dialect=True,
)
def test_explicit_join_6(self):
Node = self.classes.Node
sess = fixture_session()
n1 = aliased(Node)
node = (
sess.query(Node)
.select_from(join(Node, n1, n1.children))
.filter(n1.data == "n122")
.first()
)
assert node.data == "n12"
def test_explicit_join_7(self):
Node = self.classes.Node
sess = fixture_session()
n1 = aliased(Node)
n2 = aliased(Node)
node = (
sess.query(Node)
.select_from(join(Node, n1, Node.children).join(n2, n1.children))
.filter(n2.data == "n122")
.first()
)
assert node.data == "n1"
def test_explicit_join_8(self):
Node = self.classes.Node
sess = fixture_session()
n1 = aliased(Node)
n2 = aliased(Node)
node = (
sess.query(Node)
.select_from(
join(Node, n1, Node.id == n1.parent_id).join(n2, n1.children)
)
.filter(n2.data == "n122")
.first()
)
assert node.data == "n1"
def test_explicit_join_9(self):
Node = self.classes.Node
sess = fixture_session()
n1 = aliased(Node)
n2 = aliased(Node)
node = (
sess.query(Node)
.select_from(join(Node, n1, Node.parent).join(n2, n1.parent))
.filter(
and_(Node.data == "n122", n1.data == "n12", n2.data == "n1")
)
.first()
)
assert node.data == "n122"
def test_explicit_join_10(self):
Node = self.classes.Node
sess = fixture_session()
n1 = aliased(Node)
n2 = aliased(Node)
eq_(
list(
sess.query(Node)
.select_from(join(Node, n1, Node.parent).join(n2, n1.parent))
.filter(
and_(
Node.data == "n122", n1.data == "n12", n2.data == "n1"
)
)
.with_entities(Node.data, n1.data, n2.data)
),
[("n122", "n12", "n1")],
)
def test_join_to_nonaliased(self):
Node = self.classes.Node
sess = fixture_session()
n1 = aliased(Node)
eq_(
sess.query(n1).join(n1.parent).filter(Node.data == "n1").all(),
[
Node(parent_id=1, data="n11", id=2),
Node(parent_id=1, data="n12", id=3),
Node(parent_id=1, data="n13", id=4),
],
)
eq_(
sess.query(n1)
.join(Node, n1.parent)
.filter(Node.data == "n1")
.all(),
[
Node(parent_id=1, data="n11", id=2),
Node(parent_id=1, data="n12", id=3),
Node(parent_id=1, data="n13", id=4),
],
)
def test_multiple_explicit_entities_one(self):
Node = self.classes.Node
sess = fixture_session()
parent = aliased(Node)
grandparent = aliased(Node)
eq_(
sess.query(Node, parent, grandparent)
.join(parent, Node.parent)
.join(grandparent, parent.parent)
.filter(Node.data == "n122")
.filter(parent.data == "n12")
.filter(grandparent.data == "n1")
.first(),
(Node(data="n122"), Node(data="n12"), Node(data="n1")),
)
def test_multiple_explicit_entities_two(self):
Node = self.classes.Node
sess = fixture_session()
parent = aliased(Node)
grandparent = aliased(Node)
subq = (
sess.query(Node, parent, grandparent)
.join(parent, Node.parent)
.join(grandparent, parent.parent)
.filter(Node.data == "n122")
.filter(parent.data == "n12")
.filter(grandparent.data == "n1")
.subquery()
)
na = aliased(Node, subq)
pa = aliased(parent, subq)
ga = aliased(grandparent, subq)
eq_(
sess.query(na, pa, ga).first(),
(Node(data="n122"), Node(data="n12"), Node(data="n1")),
)
def test_multiple_explicit_entities_three(self):
Node = self.classes.Node
sess = fixture_session()
parent = aliased(Node)
grandparent = aliased(Node)
subq = (
sess.query(parent, grandparent, Node)
.join(parent, Node.parent)
.join(grandparent, parent.parent)
.filter(Node.data == "n122")
.filter(parent.data == "n12")
.filter(grandparent.data == "n1")
.subquery()
)
na = aliased(Node, subq)
pa = aliased(parent, subq)
ga = aliased(grandparent, subq)
eq_(
sess.query(pa, ga, na).first(),
(Node(data="n12"), Node(data="n1"), Node(data="n122")),
)
def test_multiple_explicit_entities_four(self):
Node = self.classes.Node
sess = fixture_session()
parent = aliased(Node)
grandparent = aliased(Node)
eq_(
sess.query(Node, parent, grandparent)
.join(parent, Node.parent)
.join(grandparent, parent.parent)
.filter(Node.data == "n122")
.filter(parent.data == "n12")
.filter(grandparent.data == "n1")
.options(joinedload(Node.children))
.first(),
(Node(data="n122"), Node(data="n12"), Node(data="n1")),
)
def test_multiple_explicit_entities_five(self):
Node = self.classes.Node
sess = fixture_session()
parent = aliased(Node)
grandparent = aliased(Node)
subq = (
sess.query(Node, parent, grandparent)
.join(parent, Node.parent)
.join(grandparent, parent.parent)
.filter(Node.data == "n122")
.filter(parent.data == "n12")
.filter(grandparent.data == "n1")
.subquery()
)
na = aliased(Node, subq)
pa = aliased(parent, subq)
ga = aliased(grandparent, subq)
eq_(
sess.query(na, pa, ga).options(joinedload(na.children)).first(),
(Node(data="n122"), Node(data="n12"), Node(data="n1")),
)
def test_any(self):
Node = self.classes.Node
sess = fixture_session()
eq_(
sess.query(Node)
.filter(Node.children.any(Node.data == "n1"))
.all(),
[],
)
eq_(
sess.query(Node)
.filter(Node.children.any(Node.data == "n12"))
.all(),
[Node(data="n1")],
)
eq_(
sess.query(Node)
.filter(~Node.children.any())
.order_by(Node.id)
.all(),
[
Node(data="n11"),
Node(data="n13"),
Node(data="n121"),
Node(data="n122"),
Node(data="n123"),
],
)
def test_has(self):
Node = self.classes.Node
sess = fixture_session()
eq_(
sess.query(Node)
.filter(Node.parent.has(Node.data == "n12"))
.order_by(Node.id)
.all(),
[Node(data="n121"), Node(data="n122"), Node(data="n123")],
)
eq_(
sess.query(Node)
.filter(Node.parent.has(Node.data == "n122"))
.all(),
[],
)
eq_(
sess.query(Node).filter(~Node.parent.has()).all(),
[Node(data="n1")],
)
def test_contains(self):
Node = self.classes.Node
sess = fixture_session()
n122 = sess.query(Node).filter(Node.data == "n122").one()
eq_(
sess.query(Node).filter(Node.children.contains(n122)).all(),
[Node(data="n12")],
)
n13 = sess.query(Node).filter(Node.data == "n13").one()
eq_(
sess.query(Node).filter(Node.children.contains(n13)).all(),
[Node(data="n1")],
)
def test_eq_ne(self):
Node = self.classes.Node
sess = fixture_session()
n12 = sess.query(Node).filter(Node.data == "n12").one()
eq_(
sess.query(Node).filter(Node.parent == n12).all(),
[Node(data="n121"), Node(data="n122"), Node(data="n123")],
)
eq_(
sess.query(Node).filter(Node.parent != n12).all(),
[
Node(data="n1"),
Node(data="n11"),
Node(data="n12"),
Node(data="n13"),
],
)
class SelfReferentialM2MTest(fixtures.MappedTest):
run_setup_mappers = "once"
run_inserts = "once"
run_deletes = None
@classmethod
def define_tables(cls, metadata):
Table(
"nodes",
metadata,
Column(
"id", Integer, primary_key=True, test_needs_autoincrement=True
),
Column("data", String(30)),
)
Table(
"node_to_nodes",
metadata,
Column(
"left_node_id",
Integer,
ForeignKey("nodes.id"),
primary_key=True,
),
Column(
"right_node_id",
Integer,
ForeignKey("nodes.id"),
primary_key=True,
),
)
@classmethod
def setup_classes(cls):
class Node(cls.Comparable):
pass
@classmethod
def insert_data(cls, connection):
Node, nodes, node_to_nodes = (
cls.classes.Node,
cls.tables.nodes,
cls.tables.node_to_nodes,
)
cls.mapper_registry.map_imperatively(
Node,
nodes,
properties={
"children": relationship(
Node,
lazy="select",
secondary=node_to_nodes,
primaryjoin=nodes.c.id == node_to_nodes.c.left_node_id,
secondaryjoin=nodes.c.id == node_to_nodes.c.right_node_id,
)
},
)
sess = Session(connection)
n1 = Node(data="n1")
n2 = Node(data="n2")
n3 = Node(data="n3")
n4 = Node(data="n4")
n5 = Node(data="n5")
n6 = Node(data="n6")
n7 = Node(data="n7")
n1.children = [n2, n3, n4]
n2.children = [n3, n6, n7]
n3.children = [n5, n4]
sess.add(n1)
sess.add(n2)
sess.add(n3)
sess.add(n4)
sess.flush()
sess.close()
def test_any(self):
Node = self.classes.Node
sess = fixture_session()
eq_(
sess.query(Node)
.filter(Node.children.any(Node.data == "n3"))
.order_by(Node.data)
.all(),
[Node(data="n1"), Node(data="n2")],
)
def test_contains(self):
Node = self.classes.Node
sess = fixture_session()
n4 = sess.query(Node).filter_by(data="n4").one()
eq_(
sess.query(Node)
.filter(Node.children.contains(n4))
.order_by(Node.data)
.all(),
[Node(data="n1"), Node(data="n3")],
)
eq_(
sess.query(Node)
.filter(not_(Node.children.contains(n4)))
.order_by(Node.data)
.all(),
[
Node(data="n2"),
Node(data="n4"),
Node(data="n5"),
Node(data="n6"),
Node(data="n7"),
],
)
def test_explicit_join(self):
Node = self.classes.Node
sess = fixture_session()
n1 = aliased(Node)
eq_(
sess.query(Node)
.select_from(join(Node, n1, Node.children))
.filter(n1.data.in_(["n3", "n7"]))
.order_by(Node.id)
.all(),
[Node(data="n1"), Node(data="n2")],
)
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)
def test_select_subquery(self):
Person, Book = self.classes("Person", "Book")
s = fixture_session()
subq = (
s.query(Book.book_id)
.correlate(Person)
.filter(Person.people_id == Book.book_owner_id)
.subquery()
.lateral()
)
stmt = s.query(Person, subq.c.book_id).join(subq, true())
self.assert_compile(
stmt,
"SELECT people.people_id AS people_people_id, "
"people.age AS people_age, people.name AS people_name, "
"anon_1.book_id AS anon_1_book_id "
"FROM people JOIN LATERAL "
"(SELECT books.book_id AS book_id FROM books "
"WHERE people.people_id = books.book_owner_id) AS anon_1 ON true",
)
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(pa.people_id == Book.book_owner_id)
.subquery()
.lateral()
)
stmt = s.query(pa, subq.c.book_id).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_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(pa.people_id == Book.book_owner_id)
.subquery()
.lateral()
)
stmt = s.query(pa, subq.c.book_id).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_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(pa)
.where(pa.people_id == Book.book_owner_id)
.subquery()
.lateral()
)
stmt = s.query(pa, subq.c.book_id).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_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(pa)
.filter(pa.people_id == Book.book_owner_id)
.subquery()
.lateral()
)
stmt = s.query(pa, subq.c.book_id).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(self):
Bookcase = self.classes.Bookcase
s = fixture_session()
srf = lateral(func.generate_series(1, Bookcase.bookcase_shelves))
self.assert_compile(
s.query(Bookcase).join(srf, true()),
"SELECT bookcases.bookcase_id AS bookcases_bookcase_id, "
"bookcases.bookcase_owner_id AS bookcases_bookcase_owner_id, "
"bookcases.bookcase_shelves AS bookcases_bookcase_shelves, "
"bookcases.bookcase_width AS bookcases_bookcase_width "
"FROM bookcases JOIN "
"LATERAL generate_series(:generate_series_1, "
"bookcases.bookcase_shelves) AS anon_1 ON true",
)
def test_from_function_aas(self):
Bookcase = self.classes.Bookcase
s = fixture_session()
subq = s.query(Bookcase).subquery()
ba = aliased(Bookcase, subq)
srf = lateral(func.generate_series(1, ba.bookcase_shelves))
self.assert_compile(
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, "
"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",
)
class JoinRawTablesWLegacyTest(QueryTest, AssertsCompiledSQL):
"""test issue 6003 where creating a legacy query with only Core elements
fails to accommodate for the ORM context thus producing a query
that ignores the "legacy" joins
"""
__dialect__ = "default"
@testing.combinations(
(
lambda sess, User, Address: sess.query(User).join(Address),
"SELECT users.id AS users_id, users.name AS users_name FROM "
"users JOIN addresses ON users.id = addresses.user_id",
),
(
lambda sess, user_table, address_table: sess.query(
user_table
).join(address_table),
"SELECT users.id AS users_id, users.name AS users_name FROM "
"users JOIN addresses ON users.id = addresses.user_id",
),
(
lambda sess, User, Address, Order: sess.query(User)
.outerjoin(Order)
.join(Address),
"SELECT users.id AS users_id, users.name AS users_name FROM "
"users LEFT OUTER JOIN orders ON users.id = orders.user_id "
"JOIN addresses ON addresses.id = orders.address_id",
),
(
lambda sess, user_table, address_table, order_table: sess.query(
user_table
)
.outerjoin(order_table)
.join(address_table),
"SELECT users.id AS users_id, users.name AS users_name FROM "
"users LEFT OUTER JOIN orders ON users.id = orders.user_id "
"JOIN addresses ON addresses.id = orders.address_id",
),
)
def test_join_render(self, spec, expected):
User, Address, Order = self.classes("User", "Address", "Order")
user_table, address_table, order_table = self.tables(
"users", "addresses", "orders"
)
sess = fixture_session()
q = testing.resolve_lambda(spec, **locals())
self.assert_compile(q, expected)
self.assert_compile(
q.set_label_style(LABEL_STYLE_TABLENAME_PLUS_COL).statement,
expected,
)
def test_core_round_trip(self):
user_table, address_table = self.tables("users", "addresses")
sess = fixture_session()
q = (
sess.query(user_table)
.join(address_table)
.where(address_table.c.email_address.startswith("ed"))
)
eq_(q.all(), [(8, "ed"), (8, "ed"), (8, "ed")])