def attr(self):
"""Return a tuple of ``(local_attr, remote_attr)``.
- This attribute is convenient when specifying a join
- using :meth:`_query.Query.join` across two relationships::
+ This attribute was originally intended to facilitate using the
+ :meth:`_query.Query.join` method to join across the two relationships
+ at once, however this makes use of a deprecated calling style.
+
+ To use :meth:`_sql.select.join` or :meth:`_orm.Query.join` with
+ an association proxy, the current method is to make use of the
+ :attr:`.AssociationProxyInstance.local_attr` and
+ :attr:`.AssociationProxyInstance.remote_attr` attributes separately::
+
+ stmt = (
+ select(Parent).
+ join(Parent.proxied.local_attr).
+ join(Parent.proxied.remote_attr)
+ )
- sess.query(Parent).join(*Parent.proxied.attr)
+ A future release may seek to provide a more succinct join pattern
+ for association proxy attributes.
.. seealso::
# "enable" options, to turn on the properties that we want to
# load by default (subject to options from the query)
enable_opt.set_generic_strategy(
- (prop.key,), dict(prop.strategy_key)
+ # convert string name to an attribute before passing
+ # to loader strategy
+ (getattr(entity.entity_namespace, prop.key),),
+ dict(prop.strategy_key),
)
else:
# "disable" options, to turn off the properties from the
# superclass that we *don't* want to load, applied after
# the options from the query to override them
disable_opt.set_generic_strategy(
- (prop.key,), {"do_nothing": True}
+ # convert string name to an attribute before passing
+ # to loader strategy
+ (getattr(entity.entity_namespace, prop.key),),
+ {"do_nothing": True},
)
primary_key = [
)
if isinstance(attr, util.string_types):
+
default_token = attr.endswith(_DEFAULT_TOKEN)
attr_str_name = attr
if attr.endswith(_WILDCARD_TOKEN) or default_token:
"Using strings to indicate column or "
"relationship paths in loader options is deprecated "
"and will be removed in SQLAlchemy 2.0. Please use "
- "the class-bound attribute directly."
+ "the class-bound attribute directly.",
)
try:
# use getattr on the class to work around
r"The Query\.with_parent\(\) method",
r"The Query\.select_entity_from\(\) method",
r"The ``aliased`` and ``from_joinpoint`` keyword arguments",
- r"Using strings to indicate relationship names in Query.join",
- r"Using strings to indicate column or relationship paths in "
- "loader options",
- r"Using strings to indicate relationship names in the ORM "
- r"with_parent\(\)",
r"The Query.with_polymorphic\(\) method is considered "
"legacy as of the 1.x series",
- r"Passing a chain of multiple join conditions to Query.join\(\) "
- r"is deprecated and will be removed in SQLAlchemy 2.0.",
- r"Query.join\(\) will no longer accept tuples as arguments",
#
# ORM Session
#
where = lib
[tool:pytest]
-addopts = --tb native -v -r sfxX --maxfail=25 -p no:warnings -p no:logging
+addopts = --tb native -v -r sfxX --maxfail=250 -p no:warnings -p no:logging
python_files = test/*test_*.py
[upload]
def test_join_separate_attr(self):
User = self.classes.User
self.assert_compile(
- self.session.query(User).join(
- User.keywords.local_attr, User.keywords.remote_attr
- ),
+ self.session.query(User)
+ .join(User.keywords.local_attr)
+ .join(User.keywords.remote_attr),
"SELECT users.id AS users_id, users.name AS users_name, "
"users.singular_id AS users_singular_id "
"FROM users JOIN userkeywords ON users.id = "
def test_join_single_attr(self):
User = self.classes.User
self.assert_compile(
- self.session.query(User).join(*User.keywords.attr),
+ self.session.query(User)
+ .join(User.keywords.attr[0])
+ .join(User.keywords.attr[1]),
"SELECT users.id AS users_id, users.name AS users_name, "
"users.singular_id AS users_singular_id "
"FROM users JOIN userkeywords ON users.id = "
session.add(book)
session.commit()
- result = session.query(Book).options(selectinload("pages")).all()
+ result = session.query(Book).options(selectinload(Book.pages)).all()
eq_(result, [book])
def go():
testcar = session.get(
- Car, car1.car_id, options=[joinedload("employee")]
+ Car, car1.car_id, options=[joinedload(Car.employee)]
)
assert str(testcar.employee) == "Engineer E4, status X"
testcar = session.get(
Car,
car1.car_id,
- options=[joinedload("employee")],
+ options=[joinedload(Car.employee)],
)
assert str(testcar.employee) == "Engineer E4, status X"
session.expunge_all()
s = session.query(Car)
- c = s.join("employee").filter(Person.name == "E4")[0]
+ c = s.join(Car.employee).filter(Person.name == "E4")[0]
assert c.car_id == car1.car_id
r = (
session.query(Person)
.filter(Person.name.like("%2"))
- .join("status")
+ .join(Person.status)
.filter_by(name="active")
.order_by(Person.person_id)
)
session = fixture_session()
r = (
session.query(Engineer)
- .join("status")
+ .join(Engineer.status)
.filter(
Person.name.in_(["E2", "E3", "E4", "M4", "M2", "M1"])
& (Status.name == "active")
s.commit()
- q = s.query(B, B.ds.any(D.id == 1)).options(joinedload("es"))
+ q = s.query(B, B.ds.any(D.id == 1)).options(joinedload(B.es))
q = q.join(C, C.b_id == B.id)
q = q.limit(5)
eq_(q.all(), [(b, True)])
result = (
session.query(A)
.filter_by(child_id=None)
- .options(joinedload("child"))
+ .options(joinedload(A.child))
.one()
)
sess = fixture_session()
eq_(
sess.query(Person)
- .join("paperwork")
+ .join(Person.paperwork)
.filter(Paperwork.description.like("%review%"))
.all(),
[b1, m1],
eq_(
sess.query(Person)
.order_by(Person.person_id)
- .join("paperwork")
+ .join(Person.paperwork)
.filter(Paperwork.description.like("%#2%"))
.all(),
[e1, m1],
eq_(
sess.query(Engineer)
.order_by(Person.person_id)
- .join("paperwork")
+ .join(Person.paperwork)
.filter(Paperwork.description.like("%#2%"))
.all(),
[e1],
eq_(
sess.query(Person)
.order_by(Person.person_id)
- .join("paperwork")
+ .join(Person.paperwork)
.filter(Person.name.like("%dog%"))
.filter(Paperwork.description.like("%#2%"))
.all(),
eq_(
sess.query(Person)
.order_by(Person.person_id)
- .join("paperwork", aliased=True)
+ .join(Person.paperwork, aliased=True)
.filter(Paperwork.description.like("%review%"))
.all(),
[b1, m1],
eq_(
sess.query(Person)
.order_by(Person.person_id)
- .join(pa, "paperwork")
+ .join(pa, Person.paperwork)
.filter(pa.description.like("%review%"))
.all(),
[b1, m1],
eq_(
sess.query(Person)
.order_by(Person.person_id)
- .join("paperwork", aliased=True)
+ .join(Person.paperwork, aliased=True)
.filter(Paperwork.description.like("%#2%"))
.all(),
[e1, m1],
eq_(
sess.query(Person)
.order_by(Person.person_id)
- .join(pa, "paperwork")
+ .join(pa, Person.paperwork)
.filter(pa.description.like("%#2%"))
.all(),
[e1, m1],
eq_(
sess.query(Engineer)
.order_by(Person.person_id)
- .join("paperwork", aliased=True)
+ .join(Person.paperwork, aliased=True)
.filter(Paperwork.description.like("%#2%"))
.all(),
[e1],
eq_(
sess.query(Engineer)
.order_by(Person.person_id)
- .join(pa, "paperwork")
+ .join(pa, Person.paperwork)
.filter(pa.description.like("%#2%"))
.all(),
[e1],
eq_(
sess.query(Person)
.order_by(Person.person_id)
- .join(pa, "paperwork")
+ .join(pa, Person.paperwork)
.filter(Person.name.like("%dog%"))
.filter(pa.description.like("%#2%"))
.all(),
sess.query(Person)
.with_polymorphic(Manager)
.order_by(Person.person_id)
- .join("paperwork")
+ .join(Person.paperwork)
.filter(Paperwork.description.like("%review%"))
.all(),
[b1, m1],
sess.query(Person)
.with_polymorphic([Manager, Engineer])
.order_by(Person.person_id)
- .join("paperwork")
+ .join(Person.paperwork)
.filter(Paperwork.description.like("%#2%"))
.all(),
[e1, m1],
sess.query(Person)
.with_polymorphic([Manager, Engineer])
.order_by(Person.person_id)
- .join("paperwork")
+ .join(Person.paperwork)
.filter(Person.name.like("%dog%"))
.filter(Paperwork.description.like("%#2%"))
.all(),
eq_(
sess.query(Person)
.with_polymorphic(Manager)
- .join("paperwork", aliased=True)
+ .join(Person.paperwork, aliased=True)
.filter(Paperwork.description.like("%review%"))
.all(),
[b1, m1],
eq_(
sess.query(Person)
.with_polymorphic(Manager)
- .join(pa, "paperwork")
+ .join(pa, Person.paperwork)
.filter(pa.description.like("%review%"))
.all(),
[b1, m1],
sess.query(Person)
.with_polymorphic([Manager, Engineer])
.order_by(Person.person_id)
- .join("paperwork", aliased=True)
+ .join(Person.paperwork, aliased=True)
.filter(Paperwork.description.like("%#2%"))
.all(),
[e1, m1],
sess.query(Person)
.with_polymorphic([Manager, Engineer])
.order_by(Person.person_id)
- .join(pa, "paperwork")
+ .join(pa, Person.paperwork)
.filter(pa.description.like("%#2%"))
.all(),
[e1, m1],
sess.query(Person)
.with_polymorphic([Manager, Engineer])
.order_by(Person.person_id)
- .join(pa, "paperwork")
+ .join(pa, Person.paperwork)
.filter(Person.name.like("%dog%"))
.filter(pa.description.like("%#2%"))
.all(),
sess = fixture_session()
eq_(
sess.query(Company)
- .join("employees")
+ .join(Company.employees)
.filter(Person.name == "vlad")
.one(),
c2,
sess = fixture_session()
eq_(
sess.query(Company)
- .join("employees", aliased=True)
+ .join(Company.employees, aliased=True)
.filter(Person.name == "vlad")
.one(),
c2,
ea = aliased(Person)
eq_(
sess.query(Company)
- .join(ea, "employees")
+ .join(ea, Company.employees)
.filter(ea.name == "vlad")
.one(),
c2,
any_ = Company.employees.any(Person.name == "wally")
eq_(
sess.query(Company)
- .join("employees", aliased=True)
+ .join(Company.employees, aliased=True)
.filter(Person.name == "dilbert")
.filter(any_)
.all(),
ealias = aliased(Engineer)
eq_(
sess.query(Company)
- .join(ealias, "employees")
+ .join(ealias, Company.employees)
.filter(ealias.primary_language == "java")
.all(),
[c1],
sess = fixture_session()
eq_(
sess.query(Company)
- .join("employees")
+ .join(Company.employees)
.filter(Engineer.primary_language == "java")
.all(),
[c1],
def test_join_to_subclass_fourteen(self):
sess = fixture_session()
eq_(
- sess.query(Company).join("employees", Engineer.machines).all(),
+ sess.query(Company)
+ .join(Company.employees)
+ .join(Engineer.machines)
+ .all(),
[c1, c2],
)
sess = fixture_session()
eq_(
sess.query(Company)
- .join("employees", Engineer.machines)
+ .join(Company.employees)
+ .join(Engineer.machines)
.filter(Machine.name.ilike("%thinkpad%"))
.all(),
[c1],
sess = fixture_session()
eq_(
sess.query(Company)
- .join("employees")
+ .join(Company.employees)
.filter(Person.name.in_(["dilbert", "vlad"]))
.join(Person.paperwork)
.filter(Paperwork.description.like("%#2%"))
sess = fixture_session()
eq_(
sess.query(Company)
- .join("employees")
+ .join(Company.employees)
.filter(Person.name.in_(["dilbert", "vlad"]))
.join(Person.paperwork)
.filter(Paperwork.description.like("%#%"))
pa = aliased(Paperwork)
eq_(
sess.query(Company)
- .join(ea, "employees")
+ .join(ea, Company.employees)
.filter(ea.name.in_(["dilbert", "vlad"]))
.join(pa, ea.paperwork)
.filter(pa.description.like("%#2%"))
pa = aliased(Person)
eq_(
sess.query(Engineer)
- .join(pa, "reports_to")
+ .join(pa, Engineer.reports_to)
.filter(pa.name == "dogbert")
.first(),
Engineer(name="dilbert"),
eq_(
sess.query(Engineer)
- .join(ma, "reports_to")
+ .join(ma, Engineer.reports_to)
.filter(ma.name == "dogbert")
.first(),
Engineer(name="dilbert"),
ea = aliased(Engineer)
eq_(
sess.query(Engineer)
- .join(ea, "reports_to")
+ .join(ea, Engineer.reports_to)
.filter(ea.name == "wally")
.first(),
Engineer(name="dilbert"),
# test that the splicing of the join works here, doesn't break in
# the middle of "parent join child1"
- q = sess.query(Child1).options(joinedload("left_child2"))
+ q = sess.query(Child1).options(joinedload(Child1.left_child2))
self.assert_compile(
q.limit(1).statement,
"SELECT child1.id, parent.id AS id_1, parent.cls, "
sess.flush()
sess.expunge_all()
- query_ = sess.query(Child1).options(subqueryload("left_child2"))
+ query_ = sess.query(Child1).options(subqueryload(Child1.left_child2))
for row in query_.all():
assert row.left_child2
def go():
eq_(
- sess.query(Subparent).options(joinedload("children")).all(),
+ sess.query(Subparent)
+ .options(joinedload(Subparent.children))
+ .all(),
[p1, p2],
)
eq_(
sess.query(Subparent)
.join(Subparent.children)
- .options(contains_eager("children"))
+ .options(contains_eager(Subparent.children))
.all(),
[p1, p2],
)
def go():
eq_(
- sess.query(Subparent).options(subqueryload("children")).all(),
+ sess.query(Subparent)
+ .options(subqueryload(Subparent.children))
+ .all(),
[p1, p2],
)
s = fixture_session()
self.assert_compile(
s.query(Parent)
- .join(Parent.sub1, Sub1.sub2)
+ .join(Parent.sub1)
+ .join(Sub1.sub2)
.join(Sub2.ep1)
.join(Sub2.ep2),
"SELECT parent.id AS parent_id, parent.data AS parent_data "
sess = fixture_session()
with self.sql_execution_asserter(testing.db) as asserter:
- sess.query(Manager).options(subqueryload("stuff")).all()
+ sess.query(Manager).options(subqueryload(Manager.stuff)).all()
asserter.assert_(
CompiledSQL(
sess = fixture_session()
self.assert_compile(
- sess.query(Company, Engineer.name).outerjoin("engineers"),
+ sess.query(Company, Engineer.name).outerjoin(Company.engineers),
"SELECT companies.company_id AS companies_company_id, "
"companies.name AS companies_name, "
"employees.name AS employees_name "
sess.expunge_all()
eq_(
sess.query(Company)
- .options(joinedload("engineers"))
+ .options(joinedload(Company.engineers))
.order_by(Company.name)
.all(),
[
.label("index"),
).alias()
- partitioned_b = cls.partitioned_b = aliased(B, alias=partition)
+ cls.partitioned_b = partitioned_b = aliased(B, alias=partition)
A.partitioned_bs = relationship(
partitioned_b,
self.assert_sql_count(testing.db, go, 2)
- @testing.combinations("string", "ac_attribute", "ac_attr_w_of_type")
+ @testing.combinations("ac_attribute", "ac_attr_w_of_type")
def test_selectinload_w_joinedload_after(self, calling_style):
"""test has been enhanced to also test #7224"""
partitioned_b = self.partitioned_b
- if calling_style == "string":
- opt = selectinload(A.partitioned_bs).joinedload("cs")
- elif calling_style == "ac_attribute":
+ if calling_style == "ac_attribute":
opt = selectinload(A.partitioned_bs).joinedload(partitioned_b.cs)
elif calling_style == "ac_attr_w_of_type":
# this would have been a workaround for people who encountered
)
s = fixture_session()
- q = s.query(Thing).options(sa.orm.joinedload("category"))
+ q = s.query(Thing).options(sa.orm.joinedload(Thing.category))
result = q.select_from(
tests.outerjoin(
)
s = fixture_session()
- q = s.query(Thing).options(sa.orm.joinedload("category"))
+ q = s.query(Thing).options(sa.orm.joinedload(Thing.category))
result = q.filter(
sa.and_(
tests.c.owner_id == 1,
options.c.someoption == False, # noqa
),
)
- ).outerjoin("owner_option")
+ ).outerjoin(Thing.owner_option)
result_str = ["%d %s" % (t.id, t.category.name) for t in result]
eq_(result_str, ["1 Some Category", "3 Some Category"])
Thing, tests = (self.classes.Thing, self.tables.tests)
s = fixture_session()
- q = s.query(Thing).options(sa.orm.joinedload("category"))
+ q = s.query(Thing).options(sa.orm.joinedload(Thing.category))
result = q.filter(
(tests.c.owner_id == 1)
& text(
"options.someoption is null or options.someoption=:opt"
).bindparams(opt=False)
- ).join("owner_option")
+ ).join(Thing.owner_option)
result_str = ["%d %s" % (t.id, t.category.name) for t in result]
eq_(result_str, ["3 Some Category"])
)
s = fixture_session()
- q = s.query(Thing).options(sa.orm.joinedload("category"))
+ q = s.query(Thing).options(sa.orm.joinedload(Thing.category))
result = q.filter(
(tests.c.owner_id == 1)
& (
(options.c.someoption == None)
| (options.c.someoption == False)
) # noqa
- ).join("owner_option")
+ ).join(Thing.owner_option)
result_str = ["%d %s" % (t.id, t.category.name) for t in result]
eq_(result_str, ["3 Some Category"])
# "order by max desc" separately
q = (
session.query(Data)
- .options(sa.orm.joinedload("foo"))
+ .options(sa.orm.joinedload(Data.foo))
.select_from(
datas.join(arb_data, arb_data.c.data_id == datas.c.id)
)
q = (
sess.query(Department)
- .join("employees")
+ .join(Department.employees)
.filter(Employee.name.startswith("J"))
.distinct()
.order_by(sa.desc(Department.name))
acc = (
session.query(Account)
.options(
- sa.orm.joinedload("entries")
- .joinedload("transaction")
- .joinedload("entries")
- .joinedload("account")
+ sa.orm.joinedload(Account.entries)
+ .joinedload(Entry.transaction)
+ .joinedload(Transaction.entries)
+ .joinedload(Entry.account)
)
.order_by(Account.account_id)
).first()
"User", "Address", "Keyword", "Order", "Item"
)
+ a1 = aliased(Address)
+
self._run_cache_key_fixture(
lambda: (
Load(User).joinedload(User.addresses),
User.orders.and_(Order.description != "somename")
),
Load(User).defer(User.id),
- Load(User).subqueryload("addresses"),
- Load(Address).defer("id"),
+ Load(User).subqueryload(User.addresses),
+ Load(Address).defer(Address.id),
Load(Address).defer("*"),
- Load(aliased(Address)).defer("id"),
+ Load(a1).defer(a1.id),
Load(User).joinedload(User.addresses).defer(Address.id),
Load(User).joinedload(User.orders).joinedload(Order.items),
Load(User).joinedload(User.orders).subqueryload(Order.items),
Load(User).defaultload(User.orders).defaultload(Order.items),
Load(User).defaultload(User.orders),
Load(Address).raiseload("*"),
- Load(Address).raiseload("user"),
+ Load(Address).raiseload(Address.user),
),
compare_values=True,
)
- def test_bound_options_equiv_on_strname(self):
- """Bound loader options resolve on string name so test that the cache
- key for the string version matches the resolved version.
-
- """
- User, Address, Keyword, Order, Item = self.classes(
- "User", "Address", "Keyword", "Order", "Item"
- )
-
- for left, right in [
- (Load(User).defer(User.id), Load(User).defer("id")),
- (
- Load(User).joinedload(User.addresses),
- Load(User).joinedload("addresses"),
- ),
- (
- Load(User).joinedload(User.orders).joinedload(Order.items),
- Load(User).joinedload("orders").joinedload("items"),
- ),
- ]:
- eq_(left._generate_cache_key(), right._generate_cache_key())
-
def test_selects_w_orm_joins(self):
User, Address, Keyword, Order, Item = self.classes(
.join(User.orders),
fixture_session()
.query(User)
- .join("addresses")
- .join("dingalings", from_joinpoint=True),
- fixture_session().query(User).join("addresses"),
- fixture_session().query(User).join("orders"),
- fixture_session().query(User).join("addresses").join("orders"),
+ .join(User.addresses)
+ .join(Address.dingaling),
fixture_session().query(User).join(Address, User.addresses),
- fixture_session().query(User).join(a1, "addresses"),
- fixture_session()
- .query(User)
- .join(a1, "addresses", aliased=True),
+ fixture_session().query(User).join(a1, User.addresses),
fixture_session().query(User).join(User.addresses.of_type(a1)),
),
compare_values=True,
def go():
g2 = (
- sess.query(Graph).options(sa.orm.joinedload("edges")).get(g.id)
+ sess.query(Graph)
+ .options(sa.orm.joinedload(Graph.edges))
+ .get(g.id)
)
eq_(
import sqlalchemy as sa
from sqlalchemy import testing
from sqlalchemy import util
+from sqlalchemy.orm import defaultload
+from sqlalchemy.orm import joinedload
from sqlalchemy.orm import relationship
+from sqlalchemy.orm import subqueryload
from sqlalchemy.testing import assert_raises_message
from sqlalchemy.testing import eq_
from sqlalchemy.testing.fixtures import fixture_session
def go():
users[:] = (
sess.query(self.classes.User)
- .options(sa.orm.subqueryload("*"))
- .options(sa.orm.joinedload(self.classes.User.addresses))
+ .options(subqueryload("*"))
+ .options(joinedload(self.classes.User.addresses))
.options(sa.orm.lazyload("*"))
.order_by(self.classes.User.id)
.all()
def test_star_must_be_alone(self):
sess = self._downgrade_fixture()
User = self.classes.User
- opt = sa.orm.subqueryload("*", User.addresses)
+ opt = subqueryload("*", User.addresses)
assert_raises_message(
sa.exc.ArgumentError,
"Wildcard token cannot be followed by another entity",
users[:] = (
sess.query(self.classes.User)
.options(sa.orm.lazyload("*"))
- .options(sa.orm.joinedload(self.classes.User.addresses))
+ .options(joinedload(self.classes.User.addresses))
.order_by(self.classes.User.id)
.all()
)
users[:] = (
sess.query(self.classes.User)
.options(sa.orm.lazyload("*"))
- .options(sa.orm.subqueryload(self.classes.User.orders))
+ .options(subqueryload(self.classes.User.orders))
.order_by(self.classes.User.id)
.all()
)
users[:] = (
sess.query(self.classes.User)
.options(sa.orm.noload("*"))
- .options(sa.orm.joinedload(self.classes.User.addresses))
+ .options(joinedload(self.classes.User.addresses))
.order_by(self.classes.User.id)
.all()
)
users[:] = (
sess.query(self.classes.User)
.options(sa.orm.noload("*"))
- .options(sa.orm.subqueryload(self.classes.User.orders))
+ .options(subqueryload(self.classes.User.orders))
.order_by(self.classes.User.id)
.all()
)
def go():
users[:] = (
sess.query(self.classes.User)
- .options(sa.orm.joinedload("*"))
+ .options(joinedload("*"))
.order_by(self.classes.User.id)
.all()
)
sess = self._upgrade_fixture()
users = []
+ User, Order, Item = self.classes("User", "Order", "Item")
+
# test upgrade all to joined: 1 sql
def go():
users[:] = (
- sess.query(self.classes.User)
- .options(sa.orm.joinedload(".*"))
- .options(sa.orm.joinedload("addresses.*"))
- .options(sa.orm.joinedload("orders.*"))
- .options(sa.orm.joinedload("orders.items.*"))
+ sess.query(User)
+ .options(joinedload(".*"))
+ .options(defaultload(User.addresses).joinedload("*"))
+ .options(defaultload(User.orders).joinedload("*"))
+ .options(
+ defaultload(User.orders)
+ .defaultload(Order.items)
+ .joinedload("*")
+ )
.order_by(self.classes.User.id)
.all()
)
sess = self._upgrade_fixture()
users = []
+ User, Order, Item = self.classes("User", "Order", "Item")
+
# test joined all but 'keywords': upgraded to 1 sql
def go():
users[:] = (
- sess.query(self.classes.User)
- .options(sa.orm.lazyload("orders.items.keywords"))
- .options(sa.orm.joinedload("*"))
- .order_by(self.classes.User.id)
+ sess.query(User)
+ .options(
+ defaultload(User.orders)
+ .defaultload(Order.items)
+ .lazyload(Item.keywords)
+ )
+ .options(joinedload("*"))
+ .order_by(User.id)
.all()
)
def go():
users[:] = (
sess.query(self.classes.User)
- .options(sa.orm.subqueryload(self.classes.User.addresses))
- .options(sa.orm.joinedload("*"))
+ .options(subqueryload(self.classes.User.addresses))
+ .options(joinedload("*"))
.order_by(self.classes.User.id)
.all()
)
def go():
users[:] = (
sess.query(self.classes.User)
- .options(sa.orm.subqueryload("*"))
+ .options(subqueryload("*"))
.order_by(self.classes.User.id)
.all()
)
sess = self._upgrade_fixture()
users = []
+ User, Order = self.classes("User", "Order")
+
# test upgrade all to subquery: 1 sql + 4 relationships = 5
def go():
users[:] = (
- sess.query(self.classes.User)
- .options(sa.orm.subqueryload(".*"))
- .options(sa.orm.subqueryload("addresses.*"))
- .options(sa.orm.subqueryload("orders.*"))
- .options(sa.orm.subqueryload("orders.items.*"))
- .order_by(self.classes.User.id)
+ sess.query(User)
+ .options(subqueryload(".*"))
+ .options(defaultload(User.addresses).subqueryload("*"))
+ .options(defaultload(User.orders).subqueryload("*"))
+ .options(
+ defaultload(User.orders)
+ .defaultload(Order.items)
+ .subqueryload("*")
+ )
+ .order_by(User.id)
.all()
)
is still honored"""
sess = self._upgrade_fixture()
users = []
+ User, Order, Item = self.classes("User", "Order", "Item")
# test subquery all but 'keywords' (1 sql + 3 relationships = 4)
def go():
users[:] = (
- sess.query(self.classes.User)
- .options(sa.orm.lazyload("orders.items.keywords"))
- .options(sa.orm.subqueryload("*"))
- .order_by(self.classes.User.id)
+ sess.query(User)
+ .options(
+ defaultload(User.orders)
+ .defaultload(Order.items)
+ .lazyload(Item.keywords)
+ )
+ .options(subqueryload("*"))
+ .order_by(User.id)
.all()
)
def go():
users[:] = (
sess.query(self.classes.User)
- .options(sa.orm.joinedload(self.classes.User.addresses))
- .options(sa.orm.joinedload(self.classes.User.orders))
- .options(sa.orm.subqueryload("*"))
+ .options(joinedload(self.classes.User.addresses))
+ .options(joinedload(self.classes.User.orders))
+ .options(subqueryload("*"))
.order_by(self.classes.User.id)
.all()
)
)
),
)
- q = fixture_session().query(m).options(sa.orm.lazyload("addresses"))
+ q = fixture_session().query(m).options(sa.orm.lazyload(User.addresses))
result = [None]
def go():
a1 = (
s.query(Address)
.filter_by(id=1)
- .options(sa.orm.noload("user"))
+ .options(sa.orm.noload(Address.user))
.first()
)
self.mapper_registry.map_imperatively(Order, orders)
sess = fixture_session()
- q = sess.query(Order).order_by(Order.id).options(defer("user_id"))
+ q = sess.query(Order).order_by(Order.id).options(defer(Order.user_id))
def go():
q.all()[0].user_id
)
sess.expunge_all()
- q2 = q.options(undefer("user_id"))
+ q2 = q.options(undefer(Order.user_id))
self.sql_eq_(
q2.all,
[
eq_(item.description, "item 4")
sess.expunge_all()
- result = q.options(undefer("orders.items.description")).all()
+ result = q.options(
+ defaultload(User.orders)
+ .defaultload(Order.items)
+ .undefer(Item.description)
+ ).all()
item = result[0].orders[1].items[1]
def go():
sess = fixture_session()
q = sess.query(User).options(
- joinedload(User.orders).defer("description").defer("isopen")
+ joinedload(User.orders)
+ .defer(Order.description)
+ .defer(Order.isopen)
)
self.assert_compile(
q,
self.mapper_registry.map_imperatively(Order, orders)
sess = fixture_session()
- q = sess.query(Order).options(load_only("isopen", "description"))
+ q = sess.query(Order).options(
+ load_only(Order.isopen, Order.description)
+ )
self.assert_compile(
q,
"SELECT orders.id AS orders_id, "
q = (
sess.query(Order)
.order_by(Order.id)
- .options(load_only("isopen", "description"))
+ .options(load_only(Order.isopen, Order.description))
)
eq_(q.first(), Order(id=1))
sess = fixture_session()
q = sess.query(Order).options(
- load_only("isopen", "description"), undefer("user_id")
+ load_only(Order.isopen, Order.description), undefer(Order.user_id)
)
self.assert_compile(
q,
"orders.isopen AS orders_isopen FROM orders",
)
- @testing.combinations(("string",), ("attr",))
- def test_load_only_synonym(self, type_):
+ def test_load_only_synonym(self):
orders, Order = self.tables.orders, self.classes.Order
self.mapper_registry.map_imperatively(
properties={"desc": synonym("description")},
)
- if type_ == "attr":
- opt = load_only(Order.isopen, Order.desc)
- else:
- opt = load_only("isopen", "desc")
+ opt = load_only(Order.isopen, Order.desc)
sess = fixture_session()
q = sess.query(Order).options(opt)
opt = (
Load(User)
.defaultload(User.addresses)
- .load_only("id", "email_address")
+ .load_only(Address.id, Address.email_address)
)
else:
- opt = defaultload(User.addresses).load_only("id", "email_address")
+ opt = defaultload(User.addresses).load_only(
+ Address.id, Address.email_address
+ )
q = sess.query(User).options(opt).filter(User.id.in_([7, 8]))
def go():
sess = fixture_session()
q = sess.query(User, Order, Address).options(
- Load(User).load_only("name"),
- Load(Order).load_only("id"),
- Load(Address).load_only("id", "email_address"),
+ Load(User).load_only(User.name),
+ Load(Order).load_only(Order.id),
+ Load(Address).load_only(Address.id, Address.email_address),
)
self.assert_compile(
sess = fixture_session()
q = sess.query(User).options(
- load_only("name")
- .defaultload("addresses")
- .load_only("id", "email_address"),
- defaultload("orders").load_only("id"),
+ load_only(User.name)
+ .defaultload(User.addresses)
+ .load_only(Address.id, Address.email_address),
+ defaultload(User.orders).load_only(Order.id),
)
# hmmmm joinedload seems to be forcing users.id into here...
q = (
s.query(Manager)
.order_by(Manager.person_id)
- .options(load_only("status", "manager_name"))
+ .options(load_only(Manager.status, Manager.manager_name))
)
self.assert_compile(
q,
q = (
s.query(Manager)
.order_by(Manager.person_id)
- .options(Load(Manager).load_only("status", "manager_name"))
+ .options(
+ Load(Manager).load_only(Manager.status, Manager.manager_name)
+ )
)
self.assert_compile(
q,
q = (
s.query(Boss)
.order_by(Person.person_id)
- .options(load_only("status", "manager_name"))
+ .options(load_only(Boss.status, Boss.manager_name))
)
self.assert_compile(
q,
q = (
s.query(Boss)
.order_by(Person.person_id)
- .options(Load(Boss).load_only("status", "manager_name"))
+ .options(Load(Boss).load_only(Boss.status, Manager.manager_name))
)
self.assert_compile(
q,
q = (
s.query(m1)
.order_by(m1.person_id)
- .options(load_only("status", "manager_name"))
+ .options(load_only(m1.status, m1.manager_name))
)
self.assert_compile(
q,
q = (
s.query(m1)
.order_by(m1.person_id)
- .options(Load(m1).load_only("status", "manager_name"))
+ .options(Load(m1).load_only(m1.status, m1.manager_name))
)
self.assert_compile(
q,
.join(Company.managers)
.options(
contains_eager(Company.managers).load_only(
- "status", "manager_name"
+ Manager.status, Manager.manager_name
)
)
)
.options(
Load(Company)
.contains_eager(Company.managers)
- .load_only("status", "manager_name")
+ .load_only(Manager.status, Manager.manager_name)
)
)
self.assert_compile(
q = (
s.query(Manager)
.order_by(Person.person_id)
- .options(defer(".*"), undefer("status"))
+ .options(defer(".*"), undefer(Manager.status))
)
self.assert_compile(
q,
def test_load_only_subclass_of_type(self):
s = fixture_session()
q = s.query(Company).options(
- joinedload(Company.employees.of_type(Manager)).load_only("status")
+ joinedload(Company.employees.of_type(Manager)).load_only(
+ Manager.status
+ )
)
self.assert_compile(
q,
def test_defer_super_name_on_subclass(self):
s = fixture_session()
- q = s.query(Manager).order_by(Person.person_id).options(defer("name"))
+ q = (
+ s.query(Manager)
+ .order_by(Person.person_id)
+ .options(defer(Person.name))
+ )
self.assert_compile(
q,
"SELECT managers.person_id AS managers_person_id, "
q = (
s.query(Manager)
.order_by(Person.person_id)
- .options(Load(Manager).defer("name"))
+ .options(Load(Manager).defer(Manager.name))
)
self.assert_compile(
q,
wp = with_polymorphic(Person, [Manager], flat=True)
- # needs to be explicit, we don't currently dig onto all the
- # sub-entities in the wp
- assert_raises_message(
- sa.exc.ArgumentError,
- r'Can\'t find property named "status" on '
- r"with_polymorphic\(Person, \[Manager\]\) in this Query.",
- s.query(Company)
- .options(
- joinedload(Company.employees.of_type(wp)).load_only("status")
- )
- ._compile_context,
- )
-
assert_raises_message(
sa.exc.ArgumentError,
'Attribute "Manager.status" does not link from element '
Thing = self.classes.Thing
session = fixture_session()
- thing = session.query(Thing).options(sa.orm.undefer("name")).first()
+ thing = (
+ session.query(Thing).options(sa.orm.undefer(Thing.name)).first()
+ )
self._test(thing)
def test_query_twice_with_clear(self):
session = fixture_session()
result = session.query(Thing).first() # noqa
session.expunge_all()
- thing = session.query(Thing).options(sa.orm.undefer("name")).first()
+ thing = (
+ session.query(Thing).options(sa.orm.undefer(Thing.name)).first()
+ )
self._test(thing)
def test_query_twice_no_clear(self):
session = fixture_session()
result = session.query(Thing).first() # noqa
- thing = session.query(Thing).options(sa.orm.undefer("name")).first()
+ thing = (
+ session.query(Thing).options(sa.orm.undefer(Thing.name)).first()
+ )
self._test(thing)
def test_joinedload_with_clear(self):
session = fixture_session()
human = ( # noqa
- session.query(Human).options(sa.orm.joinedload("thing")).first()
+ session.query(Human)
+ .options(sa.orm.joinedload(Human.thing))
+ .first()
)
session.expunge_all()
- thing = session.query(Thing).options(sa.orm.undefer("name")).first()
+ thing = (
+ session.query(Thing).options(sa.orm.undefer(Thing.name)).first()
+ )
self._test(thing)
def test_joinedload_no_clear(self):
session = fixture_session()
human = ( # noqa
- session.query(Human).options(sa.orm.joinedload("thing")).first()
+ session.query(Human)
+ .options(sa.orm.joinedload(Human.thing))
+ .first()
+ )
+ thing = (
+ session.query(Thing).options(sa.orm.undefer(Thing.name)).first()
)
- thing = session.query(Thing).options(sa.orm.undefer("name")).first()
self._test(thing)
def test_join_with_clear(self):
session = fixture_session()
result = ( # noqa
- session.query(Human).add_entity(Thing).join("thing").first()
+ session.query(Human).add_entity(Thing).join(Human.thing).first()
)
session.expunge_all()
- thing = session.query(Thing).options(sa.orm.undefer("name")).first()
+ thing = (
+ session.query(Thing).options(sa.orm.undefer(Thing.name)).first()
+ )
self._test(thing)
def test_join_no_clear(self):
session = fixture_session()
result = ( # noqa
- session.query(Human).add_entity(Thing).join("thing").first()
+ session.query(Human).add_entity(Thing).join(Human.thing).first()
+ )
+ thing = (
+ session.query(Thing).options(sa.orm.undefer(Thing.name)).first()
)
- thing = session.query(Thing).options(sa.orm.undefer("name")).first()
self._test(thing)
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 literal_column
from sqlalchemy import MetaData
from sqlalchemy.orm import contains_eager
from sqlalchemy.orm import declarative_base
from sqlalchemy.orm import declared_attr
+from sqlalchemy.orm import defaultload
from sqlalchemy.orm import defer
from sqlalchemy.orm import deferred
from sqlalchemy.orm import eagerload
+from sqlalchemy.orm import exc as orm_exc
from sqlalchemy.orm import foreign
from sqlalchemy.orm import instrumentation
from sqlalchemy.orm import joinedload
+from sqlalchemy.orm import Load
+from sqlalchemy.orm import load_only
from sqlalchemy.orm import mapper
from sqlalchemy.orm import relation
from sqlalchemy.orm import relationship
from sqlalchemy.orm import scoped_session
+from sqlalchemy.orm import selectinload
from sqlalchemy.orm import Session
from sqlalchemy.orm import sessionmaker
+from sqlalchemy.orm import strategy_options
from sqlalchemy.orm import subqueryload
from sqlalchemy.orm import synonym
from sqlalchemy.orm import undefer
from sqlalchemy.testing.mock import Mock
from sqlalchemy.testing.schema import Column
from sqlalchemy.testing.schema import Table
+from sqlalchemy.util import pickle
from . import _fixtures
from .inheritance import _poly_fixtures
+from .inheritance._poly_fixtures import _Polymorphic
+from .inheritance._poly_fixtures import Company
+from .inheritance._poly_fixtures import Engineer
+from .test_ac_relationships import PartitionByFixture
from .test_bind import GetBindTest as _GetBindTest
from .test_dynamic import _DynamicFixture
from .test_events import _RemoveListeners
from .test_options import PathTest as OptionsPathTest
+from .test_options import PathTest
+from .test_options import QueryTest as OptionsQueryTest
from .test_query import QueryTest
from .test_transaction import _LocalFixture
+from ..sql.test_compare import CacheKeyFixture
join_aliased_dep = (
r"Passing a chain of multiple join conditions to Query.join\(\)"
)
+undefer_needs_chaining = (
+ r"The \*addl_attrs on orm.(?:un)?defer is deprecated. "
+ "Please use method chaining"
+)
+
join_strings_dep = "Using strings to indicate relationship names in Query.join"
join_tuple_form = (
r"Query.join\(\) will no longer accept tuples as "
"arguments in SQLAlchemy 2.0."
)
+opt_strings_dep = (
+ "Using strings to indicate column or relationship "
+ "paths in loader options"
+)
+
+wparent_strings_dep = (
+ r"Using strings to indicate relationship names "
+ r"in the ORM with_parent\(\) function"
+)
+
def _aliased_join_warning(arg=None):
return testing.expect_warnings(
)
+class CustomJoinTest(QueryTest):
+ run_setup_mappers = None
+
+ def test_double_same_mappers_flag_alias(self):
+ """test aliasing of joins with a custom join condition"""
+
+ (
+ addresses,
+ items,
+ order_items,
+ orders,
+ Item,
+ User,
+ Address,
+ Order,
+ users,
+ ) = (
+ self.tables.addresses,
+ self.tables.items,
+ self.tables.order_items,
+ self.tables.orders,
+ self.classes.Item,
+ self.classes.User,
+ self.classes.Address,
+ self.classes.Order,
+ self.tables.users,
+ )
+
+ self.mapper_registry.map_imperatively(Address, addresses)
+ self.mapper_registry.map_imperatively(
+ Order,
+ orders,
+ properties={
+ "items": relationship(
+ Item,
+ secondary=order_items,
+ lazy="select",
+ order_by=items.c.id,
+ )
+ },
+ )
+ self.mapper_registry.map_imperatively(Item, items)
+ self.mapper_registry.map_imperatively(
+ User,
+ users,
+ properties=dict(
+ addresses=relationship(Address, lazy="select"),
+ open_orders=relationship(
+ Order,
+ primaryjoin=and_(
+ orders.c.isopen == 1, users.c.id == orders.c.user_id
+ ),
+ lazy="select",
+ viewonly=True,
+ ),
+ closed_orders=relationship(
+ Order,
+ primaryjoin=and_(
+ orders.c.isopen == 0, users.c.id == orders.c.user_id
+ ),
+ lazy="select",
+ viewonly=True,
+ ),
+ ),
+ )
+ q = fixture_session().query(User)
+
+ with assertions.expect_deprecated_20(
+ join_aliased_dep,
+ join_strings_dep,
+ join_chain_dep,
+ raise_on_any_unexpected=True,
+ ):
+ eq_(
+ q.join("open_orders", "items", aliased=True)
+ .filter(Item.id == 4)
+ .join("closed_orders", "items", aliased=True)
+ .filter(Item.id == 3)
+ .all(),
+ [User(id=7)],
+ )
+
+
+class PickleTest(fixtures.MappedTest):
+ @classmethod
+ def define_tables(cls, metadata):
+ Table(
+ "users",
+ metadata,
+ Column(
+ "id", Integer, primary_key=True, test_needs_autoincrement=True
+ ),
+ Column("name", String(30), nullable=False),
+ test_needs_acid=True,
+ test_needs_fk=True,
+ )
+
+ Table(
+ "addresses",
+ metadata,
+ Column(
+ "id", Integer, primary_key=True, test_needs_autoincrement=True
+ ),
+ Column("user_id", None, ForeignKey("users.id")),
+ Column("email_address", String(50), nullable=False),
+ test_needs_acid=True,
+ test_needs_fk=True,
+ )
+ Table(
+ "orders",
+ metadata,
+ Column(
+ "id", Integer, primary_key=True, test_needs_autoincrement=True
+ ),
+ Column("user_id", None, ForeignKey("users.id")),
+ Column("address_id", None, ForeignKey("addresses.id")),
+ Column("description", String(30)),
+ Column("isopen", Integer),
+ test_needs_acid=True,
+ test_needs_fk=True,
+ )
+ Table(
+ "dingalings",
+ metadata,
+ Column(
+ "id", Integer, primary_key=True, test_needs_autoincrement=True
+ ),
+ Column("address_id", None, ForeignKey("addresses.id")),
+ Column("data", String(30)),
+ test_needs_acid=True,
+ test_needs_fk=True,
+ )
+
+ def _option_test_fixture(self):
+ users, addresses, dingalings = (
+ self.tables.users,
+ self.tables.addresses,
+ self.tables.dingalings,
+ )
+
+ # these must be module level for pickling
+ from .test_pickled import User, Address, Dingaling
+
+ self.mapper_registry.map_imperatively(
+ User,
+ users,
+ properties={"addresses": relationship(Address, backref="user")},
+ )
+ self.mapper_registry.map_imperatively(
+ Address,
+ addresses,
+ properties={"dingaling": relationship(Dingaling)},
+ )
+ self.mapper_registry.map_imperatively(Dingaling, dingalings)
+ sess = fixture_session()
+ u1 = User(name="ed")
+ u1.addresses.append(Address(email_address="ed@bar.com"))
+ sess.add(u1)
+ sess.flush()
+ sess.expunge_all()
+ return sess, User, Address, Dingaling
+
+ @testing.requires.non_broken_pickle
+ def test_became_bound_options(self):
+ sess, User, Address, Dingaling = self._option_test_fixture()
+
+ for opt in [
+ sa.orm.joinedload("addresses"),
+ sa.orm.defer("name"),
+ sa.orm.joinedload("addresses").joinedload(Address.dingaling),
+ ]:
+ with assertions.expect_deprecated_20(opt_strings_dep):
+ context = sess.query(User).options(opt)._compile_context()
+ opt = [
+ v
+ for v in context.attributes.values()
+ if isinstance(v, sa.orm.Load)
+ ][0]
+
+ opt2 = pickle.loads(pickle.dumps(opt))
+ eq_(opt.path, opt2.path)
+ eq_(opt.local_opts, opt2.local_opts)
+
+ u1 = sess.query(User).options(opt).first()
+ pickle.loads(pickle.dumps(u1))
+
+ @testing.requires.non_broken_pickle
+ @testing.combinations(
+ lambda: sa.orm.joinedload("addresses"),
+ lambda: sa.orm.defer("name"),
+ lambda Address: sa.orm.joinedload("addresses").joinedload(
+ Address.dingaling
+ ),
+ lambda: sa.orm.joinedload("addresses").raiseload("*"),
+ )
+ def test_unbound_options(self, test_case):
+ sess, User, Address, Dingaling = self._option_test_fixture()
+
+ opt = testing.resolve_lambda(test_case, User=User, Address=Address)
+ opt2 = pickle.loads(pickle.dumps(opt))
+ eq_(opt.path, opt2.path)
+
+ with assertions.expect_deprecated_20(opt_strings_dep):
+ u1 = sess.query(User).options(opt).first()
+ pickle.loads(pickle.dumps(u1))
+
+ @testing.requires.non_broken_pickle
+ @testing.combinations(
+ lambda User: sa.orm.Load(User).joinedload("addresses"),
+ lambda User: sa.orm.Load(User).joinedload("addresses").raiseload("*"),
+ lambda User: sa.orm.Load(User).defer("name"),
+ lambda User, Address: sa.orm.Load(User)
+ .joinedload("addresses")
+ .joinedload(Address.dingaling),
+ lambda User, Address: sa.orm.Load(User)
+ .joinedload("addresses", innerjoin=True)
+ .joinedload(Address.dingaling),
+ )
+ def test_bound_options(self, test_case):
+ sess, User, Address, Dingaling = self._option_test_fixture()
+
+ with assertions.expect_deprecated_20(opt_strings_dep):
+ opt = testing.resolve_lambda(test_case, User=User, Address=Address)
+
+ opt2 = pickle.loads(pickle.dumps(opt))
+ eq_(opt.path, opt2.path)
+ eq_(opt.context.keys(), opt2.context.keys())
+ eq_(opt.local_opts, opt2.local_opts)
+
+ u1 = sess.query(User).options(opt).first()
+ pickle.loads(pickle.dumps(u1))
+
+
+class SynonymTest(QueryTest, AssertsCompiledSQL):
+ __dialect__ = "default"
+
+ @classmethod
+ def setup_mappers(cls):
+ (
+ users,
+ Keyword,
+ items,
+ order_items,
+ orders,
+ Item,
+ User,
+ Address,
+ keywords,
+ Order,
+ item_keywords,
+ addresses,
+ ) = (
+ cls.tables.users,
+ cls.classes.Keyword,
+ cls.tables.items,
+ cls.tables.order_items,
+ cls.tables.orders,
+ cls.classes.Item,
+ cls.classes.User,
+ cls.classes.Address,
+ cls.tables.keywords,
+ cls.classes.Order,
+ cls.tables.item_keywords,
+ cls.tables.addresses,
+ )
+
+ cls.mapper_registry.map_imperatively(
+ User,
+ users,
+ properties={
+ "name_syn": synonym("name"),
+ "addresses": relationship(Address),
+ "orders": relationship(
+ Order, backref="user", order_by=orders.c.id
+ ), # o2m, m2o
+ "orders_syn": synonym("orders"),
+ "orders_syn_2": synonym("orders_syn"),
+ },
+ )
+ cls.mapper_registry.map_imperatively(Address, addresses)
+ cls.mapper_registry.map_imperatively(
+ Order,
+ orders,
+ properties={
+ "items": relationship(Item, secondary=order_items), # m2m
+ "address": relationship(Address), # m2o
+ "items_syn": synonym("items"),
+ },
+ )
+ cls.mapper_registry.map_imperatively(
+ Item,
+ items,
+ properties={
+ "keywords": relationship(
+ Keyword, secondary=item_keywords
+ ) # m2m
+ },
+ )
+ cls.mapper_registry.map_imperatively(Keyword, keywords)
+
+ def test_options_syn_of_syn_string(self):
+ User, Order = self.classes.User, self.classes.Order
+
+ s = fixture_session()
+
+ def go():
+ with testing.expect_deprecated_20(opt_strings_dep):
+ result = (
+ s.query(User)
+ .filter_by(name="jack")
+ .options(joinedload("orders_syn_2"))
+ .all()
+ )
+ eq_(
+ result,
+ [
+ User(
+ id=7,
+ name="jack",
+ orders=[
+ Order(description="order 1"),
+ Order(description="order 3"),
+ Order(description="order 5"),
+ ],
+ )
+ ],
+ )
+
+ self.assert_sql_count(testing.db, go, 1)
+
+
class DeprecatedQueryTest(_fixtures.FixtureTest, AssertsCompiledSQL):
__dialect__ = "default"
q1 = s.query(User).options(joinedload("addresses"))
- with testing.expect_deprecated_20(
- "Using strings to indicate column or relationship "
- "paths in loader options"
- ):
+ with testing.expect_deprecated_20(opt_strings_dep):
self.assert_compile(
q1,
"SELECT users.id AS users_id, users.name AS users_name, "
sess = fixture_session()
- with testing.expect_deprecated(
- r"The \*addl_attrs on orm.defer is deprecated. "
- "Please use method chaining"
- ):
+ with testing.expect_deprecated(undefer_needs_chaining):
sess.query(User).options(defer("addresses", "email_address"))
- with testing.expect_deprecated(
- r"The \*addl_attrs on orm.undefer is deprecated. "
- "Please use method chaining"
- ):
+ with testing.expect_deprecated(undefer_needs_chaining):
sess.query(User).options(undefer("addresses", "email_address"))
closed_mapper = User.closed_orders.entity
open_mapper = User.open_orders.entity
- eq_(
- [Order(id=1), Order(id=5)],
- fixture_session()
- .query(closed_mapper)
- .with_parent(user, property="closed_orders")
- .all(),
- )
- eq_(
- [Order(id=3)],
- fixture_session()
- .query(open_mapper)
- .with_parent(user, property="open_orders")
- .all(),
- )
+ with testing.expect_deprecated_20(wparent_strings_dep):
+ eq_(
+ [Order(id=1), Order(id=5)],
+ fixture_session()
+ .query(closed_mapper)
+ .with_parent(user, property="closed_orders")
+ .all(),
+ )
+ with testing.expect_deprecated_20(wparent_strings_dep):
+ eq_(
+ [Order(id=3)],
+ fixture_session()
+ .query(open_mapper)
+ .with_parent(user, property="open_orders")
+ .all(),
+ )
class ViewonlyFlagWarningTest(fixtures.MappedTest):
class JoinTest(QueryTest, AssertsCompiledSQL):
__dialect__ = "default"
+ @testing.combinations(
+ "string_relationship",
+ "string_relationship_only",
+ )
+ def test_filter_by_from_join(self, onclause_type):
+ User, Address = self.classes("User", "Address")
+ (address_table,) = self.tables("addresses")
+ (user_table,) = self.tables("users")
+
+ sess = fixture_session()
+ q = sess.query(User)
+
+ with assertions.expect_deprecated_20(join_strings_dep):
+ if onclause_type == "string_relationship":
+ q = q.join(Address, "addresses")
+ elif onclause_type == "string_relationship_only":
+ q = q.join("addresses")
+ 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",
+ )
+
+ 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_implicit_joins_from_aliases(self):
Item, User, Order = (
self.classes.Item,
AssertsCompiledSQL,
):
run_setup_mappers = "once"
+ __dialect__ = "default"
+
+ def test_load_only_alias_subclass(self):
+ Manager = self.classes.Manager
+
+ s = fixture_session()
+ m1 = aliased(Manager, flat=True)
+ q = (
+ s.query(m1)
+ .order_by(m1.person_id)
+ .options(load_only("status", "manager_name"))
+ )
+ with assertions.expect_deprecated_20(opt_strings_dep):
+ self.assert_compile(
+ q,
+ "SELECT managers_1.person_id AS managers_1_person_id, "
+ "people_1.person_id AS people_1_person_id, "
+ "people_1.type AS people_1_type, "
+ "managers_1.status AS managers_1_status, "
+ "managers_1.manager_name AS managers_1_manager_name "
+ "FROM people AS people_1 JOIN managers AS "
+ "managers_1 ON people_1.person_id = managers_1.person_id "
+ "ORDER BY managers_1.person_id",
+ )
+
+ def test_load_only_alias_subclass_bound(self):
+ Manager = self.classes.Manager
+
+ s = fixture_session()
+ m1 = aliased(Manager, flat=True)
+ with assertions.expect_deprecated_20(opt_strings_dep):
+ q = (
+ s.query(m1)
+ .order_by(m1.person_id)
+ .options(Load(m1).load_only("status", "manager_name"))
+ )
+ self.assert_compile(
+ q,
+ "SELECT managers_1.person_id AS managers_1_person_id, "
+ "people_1.person_id AS people_1_person_id, "
+ "people_1.type AS people_1_type, "
+ "managers_1.status AS managers_1_status, "
+ "managers_1.manager_name AS managers_1_manager_name "
+ "FROM people AS people_1 JOIN managers AS "
+ "managers_1 ON people_1.person_id = managers_1.person_id "
+ "ORDER BY managers_1.person_id",
+ )
+
+ def test_load_only_of_type_with_polymorphic(self):
+ Company, Person, Manager = self.classes("Company", "Person", "Manager")
+ s = fixture_session()
+
+ wp = with_polymorphic(Person, [Manager], flat=True)
+
+ # needs to be explicit, we don't currently dig onto all the
+ # sub-entities in the wp
+ with assertions.expect_deprecated_20(opt_strings_dep):
+ assert_raises_message(
+ sa.exc.ArgumentError,
+ r'Can\'t find property named "status" on '
+ r"with_polymorphic\(Person, \[Manager\]\) in this Query.",
+ s.query(Company)
+ .options(
+ joinedload(Company.employees.of_type(wp)).load_only(
+ "status"
+ )
+ )
+ ._compile_context,
+ )
def test_join_to_selectable(self):
people, Company, engineers, Engineer = (
# TODO: is weakref support detectable without an instance?
# self.assertRaises(
# sa.exc.ArgumentError, mapper, NoWeakrefSupport, t2)
+
+
+class DeferredOptionsTest(AssertsCompiledSQL, _fixtures.FixtureTest):
+ __dialect__ = "default"
+
+ def test_load_only_synonym(self):
+ orders, Order = self.tables.orders, self.classes.Order
+
+ self.mapper_registry.map_imperatively(
+ Order,
+ orders,
+ properties={"desc": synonym("description")},
+ )
+
+ opt = load_only("isopen", "desc")
+
+ sess = fixture_session()
+ q = sess.query(Order).options(opt)
+ with assertions.expect_deprecated_20(opt_strings_dep):
+ self.assert_compile(
+ q,
+ "SELECT orders.id AS orders_id, orders.description "
+ "AS orders_description, orders.isopen AS orders_isopen "
+ "FROM orders",
+ )
+
+ def test_deep_options(self):
+ users, items, order_items, Order, Item, User, orders = (
+ self.tables.users,
+ self.tables.items,
+ self.tables.order_items,
+ self.classes.Order,
+ self.classes.Item,
+ self.classes.User,
+ self.tables.orders,
+ )
+
+ self.mapper_registry.map_imperatively(
+ Item,
+ items,
+ properties=dict(description=deferred(items.c.description)),
+ )
+ self.mapper_registry.map_imperatively(
+ Order,
+ orders,
+ properties=dict(items=relationship(Item, secondary=order_items)),
+ )
+ self.mapper_registry.map_imperatively(
+ User,
+ users,
+ properties=dict(orders=relationship(Order, order_by=orders.c.id)),
+ )
+
+ sess = fixture_session()
+ q = sess.query(User).order_by(User.id)
+ result = q.all()
+ item = result[0].orders[1].items[1]
+
+ def go():
+ eq_(item.description, "item 4")
+
+ self.sql_count_(1, go)
+ eq_(item.description, "item 4")
+
+ sess.expunge_all()
+ with assertions.expect_deprecated(undefer_needs_chaining):
+ result = q.options(
+ undefer(User.orders, Order.items, Item.description)
+ ).all()
+ item = result[0].orders[1].items[1]
+
+ def go():
+ eq_(item.description, "item 4")
+
+ self.sql_count_(0, go)
+ eq_(item.description, "item 4")
+
+
+class OptionsTest(PathTest, OptionsQueryTest):
+ def _option_fixture(self, *arg):
+ return strategy_options._UnboundLoad._from_keys(
+ strategy_options._UnboundLoad.joinedload, arg, True, {}
+ )
+
+ def test_chained(self):
+ User = self.classes.User
+ Order = self.classes.Order
+ sess = fixture_session()
+ q = sess.query(User)
+ opt = self._option_fixture(User.orders).joinedload("items")
+ with assertions.expect_deprecated_20(opt_strings_dep):
+ self._assert_path_result(
+ opt, q, [(User, "orders"), (User, "orders", Order, "items")]
+ )
+
+ def test_chained_plus_dotted(self):
+ User = self.classes.User
+ Order = self.classes.Order
+ Item = self.classes.Item
+ sess = fixture_session()
+ q = sess.query(User)
+ opt = self._option_fixture("orders.items").joinedload("keywords")
+ with assertions.expect_deprecated_20(opt_strings_dep):
+ self._assert_path_result(
+ opt,
+ q,
+ [
+ (User, "orders"),
+ (User, "orders", Order, "items"),
+ (User, "orders", Order, "items", Item, "keywords"),
+ ],
+ )
+
+ def test_with_current_matching_string(self):
+ Item, User, Order = (
+ self.classes.Item,
+ self.classes.User,
+ self.classes.Order,
+ )
+
+ sess = fixture_session()
+ q = sess.query(Item)._with_current_path(
+ self._make_path_registry([User, "orders", Order, "items"])
+ )
+
+ opt = self._option_fixture("orders.items.keywords")
+ with assertions.expect_deprecated_20(opt_strings_dep):
+ self._assert_path_result(opt, q, [(Item, "keywords")])
+
+ def test_with_current_nonmatching_string(self):
+ Item, User, Order = (
+ self.classes.Item,
+ self.classes.User,
+ self.classes.Order,
+ )
+
+ sess = fixture_session()
+ q = sess.query(Item)._with_current_path(
+ self._make_path_registry([User, "orders", Order, "items"])
+ )
+
+ opt = self._option_fixture("keywords")
+ self._assert_path_result(opt, q, [])
+
+ opt = self._option_fixture("items.keywords")
+ self._assert_path_result(opt, q, [])
+
+ def test_path_multilevel_string(self):
+ Item, User, Order = (
+ self.classes.Item,
+ self.classes.User,
+ self.classes.Order,
+ )
+
+ sess = fixture_session()
+ q = sess.query(User)
+
+ opt = self._option_fixture("orders.items.keywords")
+ with assertions.expect_deprecated_20(opt_strings_dep):
+ self._assert_path_result(
+ opt,
+ q,
+ [
+ (User, "orders"),
+ (User, "orders", Order, "items"),
+ (User, "orders", Order, "items", Item, "keywords"),
+ ],
+ )
+
+ def test_chained_plus_multi(self):
+ User = self.classes.User
+ Order = self.classes.Order
+ Item = self.classes.Item
+ sess = fixture_session()
+ q = sess.query(User)
+ opt = self._option_fixture(User.orders, Order.items).joinedload(
+ "keywords"
+ )
+ with assertions.expect_deprecated_20(opt_strings_dep):
+ self._assert_path_result(
+ opt,
+ q,
+ [
+ (User, "orders"),
+ (User, "orders", Order, "items"),
+ (User, "orders", Order, "items", Item, "keywords"),
+ ],
+ )
+
+ def test_multi_entity_opt_on_string(self):
+ Item = self.classes.Item
+ Order = self.classes.Order
+ opt = self._option_fixture("items")
+ sess = fixture_session()
+ q = sess.query(Item, Order)
+ with assertions.expect_deprecated_20(opt_strings_dep):
+ self._assert_path_result(opt, q, [])
+
+ def test_get_path_one_level_string(self):
+ User = self.classes.User
+
+ sess = fixture_session()
+ q = sess.query(User)
+
+ opt = self._option_fixture("addresses")
+ with assertions.expect_deprecated_20(opt_strings_dep):
+ self._assert_path_result(opt, q, [(User, "addresses")])
+
+ def test_get_path_one_level_with_unrelated(self):
+ Order = self.classes.Order
+
+ sess = fixture_session()
+ q = sess.query(Order)
+ opt = self._option_fixture("addresses")
+ with assertions.expect_deprecated_20(opt_strings_dep):
+ self._assert_path_result(opt, q, [])
+
+
+class SubOptionsTest(PathTest, OptionsQueryTest):
+ run_create_tables = False
+ run_inserts = None
+ run_deletes = None
+
+ def _assert_opts(self, q, sub_opt, non_sub_opts):
+ attr_a = {}
+
+ for val in sub_opt._to_bind:
+ val._bind_loader(
+ [
+ ent.entity_zero
+ for ent in q._compile_state()._lead_mapper_entities
+ ],
+ q._compile_options._current_path,
+ attr_a,
+ False,
+ )
+
+ attr_b = {}
+
+ for opt in non_sub_opts:
+ for val in opt._to_bind:
+ val._bind_loader(
+ [
+ ent.entity_zero
+ for ent in q._compile_state()._lead_mapper_entities
+ ],
+ q._compile_options._current_path,
+ attr_b,
+ False,
+ )
+
+ for k, l in attr_b.items():
+ if not l.strategy:
+ del attr_b[k]
+
+ def strat_as_tuple(strat):
+ return (
+ strat.strategy,
+ strat.local_opts,
+ strat.propagate_to_loaders,
+ strat._of_type,
+ strat.is_class_strategy,
+ strat.is_opts_only,
+ )
+
+ eq_(
+ {path: strat_as_tuple(load) for path, load in attr_a.items()},
+ {path: strat_as_tuple(load) for path, load in attr_b.items()},
+ )
+
+ def test_invalid_two(self):
+ User, Address, Order, Item, SubItem = self.classes(
+ "User", "Address", "Order", "Item", "SubItem"
+ )
+
+ # these options are "invalid", in that User.orders -> Item.keywords
+ # is not a path. However, the "normal" option is not generating
+ # an error for now, which is bad, but we're testing here only that
+ # it works the same way, so there you go. If and when we make this
+ # case raise, then both cases should raise in the same way.
+ sub_opt = joinedload("orders").options(
+ joinedload("keywords"), joinedload("items")
+ )
+ non_sub_opts = [
+ joinedload(User.orders).joinedload(Item.keywords),
+ defaultload(User.orders).joinedload(Order.items),
+ ]
+ sess = fixture_session()
+ with assertions.expect_deprecated_20(opt_strings_dep):
+ self._assert_opts(sess.query(User), sub_opt, non_sub_opts)
+
+ def test_four_strings(self):
+ User, Address, Order, Item, SubItem, Keyword = self.classes(
+ "User", "Address", "Order", "Item", "SubItem", "Keyword"
+ )
+ sub_opt = joinedload("orders").options(
+ defer("description"),
+ joinedload("items").options(
+ joinedload("keywords").options(defer("name")),
+ defer("description"),
+ ),
+ )
+ non_sub_opts = [
+ joinedload(User.orders),
+ defaultload(User.orders).defer(Order.description),
+ defaultload(User.orders).joinedload(Order.items),
+ defaultload(User.orders)
+ .defaultload(Order.items)
+ .joinedload(Item.keywords),
+ defaultload(User.orders)
+ .defaultload(Order.items)
+ .defer(Item.description),
+ defaultload(User.orders)
+ .defaultload(Order.items)
+ .defaultload(Item.keywords)
+ .defer(Keyword.name),
+ ]
+ sess = fixture_session()
+ with assertions.expect_deprecated_20(opt_strings_dep):
+ self._assert_opts(sess.query(User), sub_opt, non_sub_opts)
+
+ def test_five_strings(self):
+ User, Address, Order, Item, SubItem, Keyword = self.classes(
+ "User", "Address", "Order", "Item", "SubItem", "Keyword"
+ )
+ sub_opt = joinedload("orders").options(load_only("description"))
+ non_sub_opts = [
+ joinedload(User.orders),
+ defaultload(User.orders).load_only(Order.description),
+ ]
+ sess = fixture_session()
+ with assertions.expect_deprecated_20(opt_strings_dep):
+ self._assert_opts(sess.query(User), sub_opt, non_sub_opts)
+
+
+class OptionsNoPropTest(_fixtures.FixtureTest):
+ """test the error messages emitted when using property
+ options in conjunction with column-only entities, or
+ for not existing options
+
+ """
+
+ run_create_tables = False
+ run_inserts = None
+ run_deletes = None
+
+ def test_option_with_column_basestring(self):
+ Item = self.classes.Item
+
+ message = (
+ "Query has only expression-based entities - can't "
+ 'find property named "keywords".'
+ )
+ self._assert_eager_with_just_column_exception(
+ Item.id, "keywords", message
+ )
+
+ def test_option_against_nonexistent_basestring(self):
+ Item = self.classes.Item
+ with assertions.expect_deprecated_20(opt_strings_dep):
+ self._assert_eager_with_entity_exception(
+ [Item],
+ (joinedload("foo"),),
+ 'Can\'t find property named "foo" on mapped class '
+ "Item->items in this Query.",
+ )
+
+ def test_option_against_nonexistent_twolevel_basestring(self):
+ Item = self.classes.Item
+ with assertions.expect_deprecated_20(opt_strings_dep):
+ self._assert_eager_with_entity_exception(
+ [Item],
+ (joinedload("keywords.foo"),),
+ 'Can\'t find property named "foo" on mapped class '
+ "Keyword->keywords in this Query.",
+ )
+
+ def test_option_against_nonexistent_twolevel_chained(self):
+ Item = self.classes.Item
+ with assertions.expect_deprecated_20(opt_strings_dep):
+ self._assert_eager_with_entity_exception(
+ [Item],
+ (joinedload("keywords").joinedload("foo"),),
+ 'Can\'t find property named "foo" on mapped class '
+ "Keyword->keywords in this Query.",
+ )
+
+ @testing.fails_if(
+ lambda: True,
+ "PropertyOption doesn't yet check for relation/column on end result",
+ )
+ def test_option_against_non_relation_basestring(self):
+ Item = self.classes.Item
+ Keyword = self.classes.Keyword
+ self._assert_eager_with_entity_exception(
+ [Keyword, Item],
+ (joinedload("keywords"),),
+ r"Attribute 'keywords' of entity 'Mapper\|Keyword\|keywords' "
+ "does not refer to a mapped entity",
+ )
+
+ @testing.fails_if(
+ lambda: True,
+ "PropertyOption doesn't yet check for relation/column on end result",
+ )
+ def test_option_against_multi_non_relation_basestring(self):
+ Item = self.classes.Item
+ Keyword = self.classes.Keyword
+ self._assert_eager_with_entity_exception(
+ [Keyword, Item],
+ (joinedload("keywords"),),
+ r"Attribute 'keywords' of entity 'Mapper\|Keyword\|keywords' "
+ "does not refer to a mapped entity",
+ )
+
+ def test_option_against_wrong_entity_type_basestring(self):
+ Item = self.classes.Item
+ with assertions.expect_deprecated_20(opt_strings_dep):
+ self._assert_loader_strategy_exception(
+ [Item],
+ (joinedload("id").joinedload("keywords"),),
+ 'Can\'t apply "joined loader" strategy to property "Item.id", '
+ 'which is a "column property"; this loader strategy is '
+ 'intended to be used with a "relationship property".',
+ )
+
+ def test_col_option_against_relationship_basestring(self):
+ Item = self.classes.Item
+ with assertions.expect_deprecated_20(opt_strings_dep):
+ self._assert_loader_strategy_exception(
+ [Item],
+ (load_only("keywords"),),
+ 'Can\'t apply "column loader" strategy to property '
+ '"Item.keywords", which is a "relationship property"; this '
+ "loader strategy is intended to be used with a "
+ '"column property".',
+ )
+
+ def test_option_against_multi_non_relation_twolevel_basestring(self):
+ Item = self.classes.Item
+ Keyword = self.classes.Keyword
+ with assertions.expect_deprecated_20(opt_strings_dep):
+ self._assert_loader_strategy_exception(
+ [Keyword, Item],
+ (joinedload("id").joinedload("keywords"),),
+ 'Can\'t apply "joined loader" strategy to property '
+ '"Keyword.id", '
+ 'which is a "column property"; this loader strategy is '
+ "intended "
+ 'to be used with a "relationship property".',
+ )
+
+ def test_option_against_multi_nonexistent_basestring(self):
+ Item = self.classes.Item
+ Keyword = self.classes.Keyword
+ with assertions.expect_deprecated_20(opt_strings_dep):
+ self._assert_eager_with_entity_exception(
+ [Keyword, Item],
+ (joinedload("description"),),
+ 'Can\'t find property named "description" on mapped class '
+ "Keyword->keywords in this Query.",
+ )
+
+ def test_option_against_multi_no_entities_basestring(self):
+ Item = self.classes.Item
+ Keyword = self.classes.Keyword
+ self._assert_eager_with_entity_exception(
+ [Keyword.id, Item.id],
+ (joinedload("keywords"),),
+ r"Query has only expression-based entities - can't find property "
+ 'named "keywords".',
+ )
+
+ def test_option_with_mapper_then_column_basestring(self):
+ Item = self.classes.Item
+
+ with assertions.expect_deprecated_20(opt_strings_dep):
+ self._assert_option([Item, Item.id], "keywords")
+
+ def test_option_with_mapper_basestring(self):
+ Item = self.classes.Item
+
+ with assertions.expect_deprecated_20(opt_strings_dep):
+ self._assert_option([Item], "keywords")
+
+ def test_option_with_column_then_mapper_basestring(self):
+ Item = self.classes.Item
+
+ with assertions.expect_deprecated_20(opt_strings_dep):
+ self._assert_option([Item.id, Item], "keywords")
+
+ @classmethod
+ def setup_mappers(cls):
+ users, User, addresses, Address, orders, Order = (
+ cls.tables.users,
+ cls.classes.User,
+ cls.tables.addresses,
+ cls.classes.Address,
+ cls.tables.orders,
+ cls.classes.Order,
+ )
+ cls.mapper_registry.map_imperatively(
+ User,
+ users,
+ properties={
+ "addresses": relationship(Address),
+ "orders": relationship(Order),
+ },
+ )
+ cls.mapper_registry.map_imperatively(Address, addresses)
+ cls.mapper_registry.map_imperatively(Order, orders)
+ keywords, items, item_keywords, Keyword, Item = (
+ cls.tables.keywords,
+ cls.tables.items,
+ cls.tables.item_keywords,
+ cls.classes.Keyword,
+ cls.classes.Item,
+ )
+ cls.mapper_registry.map_imperatively(
+ Keyword,
+ keywords,
+ properties={
+ "keywords": column_property(keywords.c.name + "some keyword")
+ },
+ )
+ cls.mapper_registry.map_imperatively(
+ Item,
+ items,
+ properties=dict(
+ keywords=relationship(Keyword, secondary=item_keywords)
+ ),
+ )
+
+ class OrderWProp(cls.classes.Order):
+ @property
+ def some_attr(self):
+ return "hi"
+
+ cls.mapper_registry.map_imperatively(
+ OrderWProp, None, inherits=cls.classes.Order
+ )
+
+ def _assert_option(self, entity_list, option):
+ Item = self.classes.Item
+
+ context = (
+ fixture_session()
+ .query(*entity_list)
+ .options(joinedload(option))
+ ._compile_state()
+ )
+ key = ("loader", (inspect(Item), inspect(Item).attrs.keywords))
+ assert key in context.attributes
+
+ def _assert_loader_strategy_exception(self, entity_list, options, message):
+ assert_raises_message(
+ orm_exc.LoaderStrategyException,
+ message,
+ fixture_session()
+ .query(*entity_list)
+ .options(*options)
+ ._compile_state,
+ )
+
+ def _assert_eager_with_entity_exception(
+ self, entity_list, options, message
+ ):
+ assert_raises_message(
+ sa.exc.ArgumentError,
+ message,
+ fixture_session()
+ .query(*entity_list)
+ .options(*options)
+ ._compile_state,
+ )
+
+ def _assert_eager_with_just_column_exception(
+ self, column, eager_option, message
+ ):
+ assert_raises_message(
+ sa.exc.ArgumentError,
+ message,
+ fixture_session()
+ .query(column)
+ .options(joinedload(eager_option))
+ ._compile_state,
+ )
+
+
+class OptionsNoPropTestInh(_Polymorphic):
+ def test_missing_str_attr_of_type_subclass(self):
+ s = fixture_session()
+
+ with assertions.expect_deprecated_20(opt_strings_dep):
+ assert_raises_message(
+ sa.exc.ArgumentError,
+ r'Can\'t find property named "manager_name" on '
+ r"mapped class Engineer->engineers in this Query.$",
+ s.query(Company)
+ .options(
+ joinedload(Company.employees.of_type(Engineer)).load_only(
+ "manager_name"
+ )
+ )
+ ._compile_state,
+ )
+
+
+class CacheKeyTest(CacheKeyFixture, _fixtures.FixtureTest):
+ """In these tests we've moved / adapted all the tests from
+ test_cache_key that make use of string options or string join(). Because
+ we are ensuring cache keys are distinct we still keep a lot of the
+ non-deprecated cases in the lists that we are testing.
+
+ """
+
+ run_setup_mappers = "once"
+ run_inserts = None
+ run_deletes = None
+
+ @classmethod
+ def setup_mappers(cls):
+ cls._setup_stock_mapping()
+
+ def _stmt_20(self, *elements):
+ return tuple(
+ elem._statement_20() if isinstance(elem, sa.orm.Query) else elem
+ for elem in elements
+ )
+
+ def _deprecated_opt(self, fn):
+ with assertions.expect_deprecated_20(
+ opt_strings_dep, raise_on_any_unexpected=True
+ ):
+ return fn()
+
+ def _deprecated_join(self, fn):
+ with assertions.expect_deprecated_20(
+ join_strings_dep, raise_on_any_unexpected=True
+ ):
+ return fn()
+
+ def _deprecated_join_w_aliased(self, fn):
+ with assertions.expect_deprecated_20(
+ join_strings_dep, join_aliased_dep, raise_on_any_unexpected=True
+ ):
+ return fn()
+
+ def test_bound_options(self):
+ User, Address, Keyword, Order, Item = self.classes(
+ "User", "Address", "Keyword", "Order", "Item"
+ )
+
+ self._run_cache_key_fixture(
+ lambda: (
+ Load(User).joinedload(User.addresses),
+ Load(User).joinedload(
+ User.addresses.of_type(aliased(Address))
+ ),
+ Load(User).joinedload(User.orders),
+ self._deprecated_opt(
+ lambda: Load(User).subqueryload("addresses")
+ ),
+ self._deprecated_opt(lambda: Load(Address).defer("id")),
+ Load(Address).defer("*"),
+ self._deprecated_opt(
+ lambda: Load(aliased(Address)).defer("id")
+ ),
+ Load(User).joinedload(User.addresses).defer(Address.id),
+ Load(User).joinedload(User.orders).joinedload(Order.items),
+ Load(User).joinedload(User.orders).subqueryload(Order.items),
+ Load(User).subqueryload(User.orders).subqueryload(Order.items),
+ Load(Address).raiseload("*"),
+ self._deprecated_opt(lambda: Load(Address).raiseload("user")),
+ ),
+ compare_values=True,
+ )
+
+ def test_bound_options_equiv_on_strname(self):
+ """Bound loader options resolve on string name so test that the cache
+ key for the string version matches the resolved version.
+
+ """
+ User, Address, Keyword, Order, Item = self.classes(
+ "User", "Address", "Keyword", "Order", "Item"
+ )
+
+ for left, right in [
+ (
+ Load(User).defer(User.id),
+ self._deprecated_opt(lambda: Load(User).defer("id")),
+ ),
+ (
+ Load(User).joinedload(User.addresses),
+ self._deprecated_opt(
+ lambda: Load(User).joinedload("addresses")
+ ),
+ ),
+ (
+ Load(User).joinedload(User.orders).joinedload(Order.items),
+ self._deprecated_opt(
+ lambda: Load(User).joinedload("orders").joinedload("items")
+ ),
+ ),
+ ]:
+ eq_(left._generate_cache_key(), right._generate_cache_key())
+
+ def test_orm_query_w_orm_joins(self):
+
+ User, Address, Keyword, Order, Item = self.classes(
+ "User", "Address", "Keyword", "Order", "Item"
+ )
+
+ a1 = aliased(Address)
+
+ self._run_cache_key_fixture(
+ lambda: self._stmt_20(
+ fixture_session().query(User).join(User.addresses),
+ fixture_session().query(User).join(User.orders),
+ fixture_session()
+ .query(User)
+ .join(User.addresses)
+ .join(User.orders),
+ self._deprecated_join_w_aliased(
+ lambda: fixture_session()
+ .query(User)
+ .join("addresses")
+ .join("dingalings", from_joinpoint=True)
+ ),
+ self._deprecated_join(
+ lambda: fixture_session().query(User).join("addresses")
+ ),
+ self._deprecated_join(
+ lambda: fixture_session().query(User).join("orders")
+ ),
+ self._deprecated_join(
+ lambda: fixture_session()
+ .query(User)
+ .join("addresses")
+ .join("orders")
+ ),
+ fixture_session().query(User).join(Address, User.addresses),
+ self._deprecated_join(
+ lambda: fixture_session().query(User).join(a1, "addresses")
+ ),
+ self._deprecated_join_w_aliased(
+ lambda: fixture_session()
+ .query(User)
+ .join(a1, "addresses", aliased=True)
+ ),
+ fixture_session().query(User).join(User.addresses.of_type(a1)),
+ ),
+ compare_values=True,
+ )
+
+ def test_unbound_options(self):
+ User, Address, Keyword, Order, Item = self.classes(
+ "User", "Address", "Keyword", "Order", "Item"
+ )
+
+ # unbound options dont emit a deprecation warning during cache
+ # key generation
+ self._run_cache_key_fixture(
+ lambda: (
+ joinedload(User.addresses),
+ joinedload(User.addresses.of_type(aliased(Address))),
+ joinedload("addresses"),
+ joinedload(User.orders),
+ joinedload(User.orders.and_(Order.id != 5)),
+ joinedload(User.orders).selectinload("items"),
+ joinedload(User.orders).selectinload(Order.items),
+ defer(User.id),
+ defer("id"),
+ defer("*"),
+ defer(Address.id),
+ joinedload(User.addresses).defer(Address.id),
+ joinedload(User.addresses).defer("id"),
+ subqueryload(User.orders)
+ .subqueryload(Order.items)
+ .defer(Item.description),
+ defaultload(User.orders).defaultload(Order.items),
+ defaultload(User.orders),
+ ),
+ compare_values=True,
+ )
+
+ def test_unbound_sub_options(self):
+ """test #6869"""
+
+ User, Address, Keyword, Order, Item = self.classes(
+ "User", "Address", "Keyword", "Order", "Item"
+ )
+
+ self._run_cache_key_fixture(
+ lambda: (
+ joinedload(User.addresses).options(
+ joinedload(Address.dingaling)
+ ),
+ joinedload(User.addresses).options(
+ joinedload(Address.dingaling).options(load_only("name"))
+ ),
+ joinedload(User.orders).options(
+ joinedload(Order.items).options(joinedload(Item.keywords))
+ ),
+ ),
+ compare_values=True,
+ )
+
+
+class AliasedClassRelationshipTest(
+ PartitionByFixture, testing.AssertsCompiledSQL
+):
+ __requires__ = ("window_functions",)
+ __dialect__ = "default"
+
+ def test_selectinload_w_joinedload_after(self):
+ """test has been enhanced to also test #7224"""
+
+ A, B, C = self.classes("A", "B", "C")
+
+ s = Session(testing.db)
+
+ opt = selectinload(A.partitioned_bs).joinedload("cs")
+
+ def go():
+ for a1 in s.query(A).options(opt):
+ for b in a1.partitioned_bs:
+ eq_(len(b.cs), 2)
+
+ with assertions.expect_deprecated_20(opt_strings_dep):
+ self.assert_sql_count(testing.db, go, 2)
def go():
eq_(
self.static.item_keyword_result[0:2],
- q.join("keywords").filter(Keyword.name == "red").all(),
+ q.join(Item.keywords).filter(Keyword.name == "red").all(),
)
self.assert_sql_count(testing.db, go, 1)
ka = aliased(Keyword)
eq_(
self.static.item_keyword_result[0:2],
- (q.join(ka, "keywords").filter(ka.name == "red")).all(),
+ (
+ q.join(Item.keywords.of_type(ka)).filter(ka.name == "red")
+ ).all(),
)
self.assert_sql_count(testing.db, go, 1)
if not testing.against("mssql"):
result = (
- q.join("orders")
+ q.join(User.orders)
.order_by(Order.user_id.desc())
.limit(2)
.offset(1)
)
result = (
- q.join("addresses")
+ q.join(User.addresses)
.order_by(Address.email_address.desc())
.limit(1)
.offset(0)
def go():
ret = (
sess.query(User, oalias)
- .join(oalias, "orders")
+ .join(User.orders.of_type(oalias))
.order_by(User.id, oalias.id)
.all()
)
ret = (
sess.query(User, oalias)
.options(joinedload(User.addresses))
- .join(oalias, "orders")
+ .join(User.orders.of_type(oalias))
.order_by(User.id, oalias.id)
.all()
)
ret = (
sess.query(User, oalias)
.options(joinedload(User.addresses), joinedload(oalias.items))
- .join(oalias, "orders")
+ .join(User.orders.of_type(oalias))
.order_by(User.id, oalias.id)
.all()
)
s.query(LDA)
.join(LDA.ld)
.options(contains_eager(LDA.ld))
- .join("a", (l_ac, "ld"), (u_ac, "user"))
+ .join(LDA.a)
+ .join(A.ld.of_type(l_ac))
+ .join(l_ac.user.of_type(u_ac))
.options(
contains_eager(LDA.a)
.contains_eager(A.ld, alias=l_ac)
assert "addresses" not in u.__dict__
(
sess.query(User)
- .options(sa.orm.joinedload("addresses"))
+ .options(sa.orm.joinedload(User.addresses))
.filter_by(id=8)
.all()
)
sess.expunge_all()
# same tests, using deferred at the options level
- o = sess.query(Order).options(sa.orm.defer("description")).get(3)
+ o = sess.query(Order).options(sa.orm.defer(Order.description)).get(3)
assert "description" not in o.__dict__
)
},
)
- q = s.query(User).options(sa.orm.lazyload("addresses"))
+ q = s.query(User).options(sa.orm.lazyload(User.addresses))
u = q.filter(users.c.id == 8).first()
def go():
q = (
s.query(User)
.filter_by(name="fred")
- .options(sa.orm.lazyload("addresses").joinedload("dingalings"))
+ .options(
+ sa.orm.lazyload(User.addresses).joinedload(Address.dingalings)
+ )
)
u1 = q.one()
def go():
result = (
- q.options(contains_eager("addresses"))
+ q.options(contains_eager(User.addresses))
.from_statement(query)
.all()
)
ulist = query.alias("ulist")
ulist_alias = aliased(User, alias=ulist)
result = (
- q.options(contains_eager("addresses", alias=ulist))
+ q.options(contains_eager(User.addresses, alias=ulist))
.select_entity_from(ulist_alias)
.all()
)
result = (
sess.query(User)
.select_entity_from(query.subquery())
- .options(contains_eager("addresses"))
+ .options(contains_eager(User.addresses))
.all()
)
assert self.static.user_address_result == result
result = (
sess.query(User)
.select_entity_from(query.subquery())
- .options(contains_eager("addresses", alias=adalias))
+ .options(contains_eager(User.addresses, alias=adalias))
.all()
)
assert self.static.user_address_result == result
def go():
result = (
- q.options(contains_eager("addresses"))
+ q.options(contains_eager(User.addresses))
.from_statement(selectquery)
.all()
)
def go():
result = (
sess.execute(
- q.options(contains_eager("addresses")).from_statement(
+ q.options(contains_eager(User.addresses)).from_statement(
selectquery
)
)
def go():
result = (
- q.options(contains_eager("addresses", alias=adalias))
+ q.options(contains_eager(User.addresses.of_type(adalias)))
.outerjoin(adalias, User.addresses)
.order_by(User.id, adalias.id)
)
self.tables.order_items,
self.classes.User,
)
+ Order = self.classes.Order
sess = fixture_session()
q = sess.query(User)
# test using Alias with more than one level deep
- # new way:
- # from sqlalchemy.orm.strategy_options import Load
- # opt = Load(User).contains_eager('orders', alias=oalias).
- # contains_eager('items', alias=ialias)
-
def go():
result = list(
q.options(
- contains_eager("orders", alias=oalias),
- contains_eager("orders.items", alias=ialias),
+ contains_eager(User.orders, alias=oalias).contains_eager(
+ Order.items, alias=ialias
+ ),
).from_statement(query)
)
assert self.static.user_order_result == result
adalias = aliased(Address)
eq_(
sess.query(User, func.count(adalias.email_address))
- .outerjoin(adalias, "addresses")
+ .outerjoin(User.addresses.of_type(adalias))
.group_by(User)
.order_by(User.id)
.all(),
results = (
sess.query(User)
.limit(1)
- .options(joinedload("addresses"))
+ .options(joinedload(User.addresses))
.add_columns(User.name)
.all()
)
q = (
sess.query(User)
.add_entity(address_entity)
- .outerjoin(address_entity, "addresses")
+ .outerjoin(address_entity, User.addresses)
.order_by(User.id, address_entity.id)
)
eq_(q.all(), expected)
sess.expunge_all()
q = sess.query(User).add_entity(address_entity)
- q = q.join(address_entity, "addresses")
+ q = q.join(address_entity, User.addresses)
q = q.filter_by(email_address="ed@bettyboop.com")
eq_(q.all(), [(user8, address3)])
sess.expunge_all()
q = (
sess.query(User, address_entity)
- .join(address_entity, "addresses")
+ .join(address_entity, User.addresses)
.filter_by(email_address="ed@bettyboop.com")
)
eq_(q.all(), [(user8, address3)])
q = (
sess.query(User, address_entity)
- .join(address_entity, "addresses")
- .options(joinedload("addresses"))
+ .join(address_entity, User.addresses)
+ .options(joinedload(User.addresses))
.filter_by(email_address="ed@bettyboop.com")
)
eq_(list(util.OrderedSet(q.all())), [(user8, address3)])
q = (
q.group_by(users)
.order_by(User.id)
- .outerjoin("addresses")
+ .outerjoin(User.addresses)
.add_columns(func.count(Address.id).label("count"))
)
eq_(q.all(), expected)
q = (
q.group_by(users)
.order_by(User.id)
- .outerjoin(adalias, "addresses")
+ .outerjoin(User.addresses.of_type(adalias))
.add_columns(func.count(adalias.id).label("count"))
)
eq_(q.all(), expected)
.add_columns(
func.count(addresses.c.id), ("Name:" + users.c.name)
)
- .outerjoin("addresses")
+ .outerjoin(User.addresses)
.group_by(users)
.order_by(users.c.id)
)
for crit, j, exp in [
(
User.id + Address.id,
- User.addresses,
+ (User.addresses,),
"SELECT users.id + addresses.id AS anon_1 "
"FROM users JOIN addresses ON users.id = "
"addresses.user_id",
),
(
User.id + Address.id,
- Address.user,
+ (Address.user,),
"SELECT users.id + addresses.id AS anon_1 "
"FROM addresses JOIN users ON users.id = "
"addresses.user_id",
),
(
Address.id + User.id,
- User.addresses,
+ (User.addresses,),
"SELECT addresses.id + users.id AS anon_1 "
"FROM users JOIN addresses ON users.id = "
"addresses.user_id",
q = s.query(crit)
mzero = q._compile_state()._entity_zero()
is_(mzero, q._compile_state()._entities[0].entity_zero)
- q = q.join(j)
+ q = q.join(*j)
self.assert_compile(q, exp)
for crit, j, exp in [
(
ua.id + Address.id,
- ua.addresses,
+ (ua.addresses,),
"SELECT users_1.id + addresses.id AS anon_1 "
"FROM users AS users_1 JOIN addresses "
"ON users_1.id = addresses.user_id",
q = s.query(crit)
mzero = q._compile_state()._entity_zero()
is_(mzero, q._compile_state()._entities[0].entity_zero)
- q = q.join(j)
+ q = q.join(*j)
self.assert_compile(q, exp)
def test_aliased_adapt_on_names(self):
eq_(
sess.query(User)
.select_entity_from(sel)
- .options(joinedload("addresses"))
+ .options(joinedload(User.addresses))
.first(),
User(name="jack", addresses=[Address(id=1)]),
)
sess = fixture_session()
self.assert_compile(
- sess.query(User).select_from(Address).join("user"),
+ sess.query(User).select_from(Address).join(Address.user),
"SELECT users.id AS users_id, users.name AS users_name "
"FROM addresses JOIN users ON users.id = addresses.user_id",
)
eq_(
sess.query(User)
.select_entity_from(sel.subquery())
- .join("addresses")
+ .join(User.addresses)
.add_entity(Address)
.order_by(User.id)
.order_by(Address.id)
eq_(
sess.query(User)
.select_entity_from(sel.subquery())
- .join(adalias, "addresses")
+ .join(adalias, User.addresses)
.add_entity(adalias)
.order_by(User.id)
.order_by(adalias.id)
eq_(
sess.query(User)
.select_entity_from(sel.subquery())
- .join(User.orders, Order.items, Item.keywords)
+ .join(User.orders)
+ .join(Order.items)
+ .join(Item.keywords)
.filter(Keyword.name.in_(["red", "big", "round"]))
.all(),
[User(name="jack", id=7)],
sess.query(User)
.select_entity_from(sel.subquery())
.options(
- joinedload("orders")
- .joinedload("items")
- .joinedload("keywords")
+ joinedload(User.orders)
+ .joinedload(Order.items)
+ .joinedload(Item.keywords)
)
- .join(User.orders, Order.items, Item.keywords)
+ .join(User.orders)
+ .join(Order.items)
+ .join(Item.keywords)
.filter(Keyword.name.in_(["red", "big", "round"]))
.all(),
[
def go():
eq_(
sess.query(User)
- .options(joinedload("addresses"))
+ .options(joinedload(User.addresses))
.select_entity_from(sel.subquery())
.order_by(User.id)
.all(),
def go():
eq_(
sess.query(User)
- .options(joinedload("addresses"))
+ .options(joinedload(User.addresses))
.select_entity_from(sel.subquery())
.filter(User.id == 8)
.order_by(User.id)
def go():
eq_(
sess.query(User)
- .options(joinedload("addresses"))
+ .options(joinedload(User.addresses))
.select_entity_from(sel.subquery())
.order_by(User.id)[1],
User(
class CustomJoinTest(QueryTest):
run_setup_mappers = None
- def test_double_same_mappers_flag_alias(self):
- """test aliasing of joins with a custom join condition"""
-
- (
- addresses,
- items,
- order_items,
- orders,
- Item,
- User,
- Address,
- Order,
- users,
- ) = (
- self.tables.addresses,
- self.tables.items,
- self.tables.order_items,
- self.tables.orders,
- self.classes.Item,
- self.classes.User,
- self.classes.Address,
- self.classes.Order,
- self.tables.users,
- )
-
- self.mapper_registry.map_imperatively(Address, addresses)
- self.mapper_registry.map_imperatively(
- Order,
- orders,
- properties={
- "items": relationship(
- Item,
- secondary=order_items,
- lazy="select",
- order_by=items.c.id,
- )
- },
- )
- self.mapper_registry.map_imperatively(Item, items)
- self.mapper_registry.map_imperatively(
- User,
- users,
- properties=dict(
- addresses=relationship(Address, lazy="select"),
- open_orders=relationship(
- Order,
- primaryjoin=and_(
- orders.c.isopen == 1, users.c.id == orders.c.user_id
- ),
- lazy="select",
- viewonly=True,
- ),
- closed_orders=relationship(
- Order,
- primaryjoin=and_(
- orders.c.isopen == 0, users.c.id == orders.c.user_id
- ),
- lazy="select",
- viewonly=True,
- ),
- ),
- )
- q = fixture_session().query(User)
-
- eq_(
- q.join("open_orders", "items", aliased=True)
- .filter(Item.id == 4)
- .join("closed_orders", "items", aliased=True)
- .filter(Item.id == 3)
- .all(),
- [User(id=7)],
- )
-
def test_double_same_mappers_explicit_alias(self):
"""test aliasing of joins with a custom join condition"""
sess = fixture_session()
- sess.query(Address).options(joinedload("user")).all()
+ sess.query(Address).options(joinedload(Address.user)).all()
eq_(
sess.query(User).all(),
def go():
eq_(
sess.query(Address)
- .options(joinedload("user"))
+ .options(joinedload(Address.user))
.order_by(Address.id)
.all(),
address_result,
ualias = aliased(User)
eq_(
- sess.query(Address, ualias).join(ualias, "user").all(),
+ sess.query(Address, ualias).join(ualias, Address.user).all(),
[(address, address.user) for address in address_result],
)
ualias2 = aliased(User)
eq_(
sess.query(Address, ualias.count)
- .join(ualias, "user")
- .join(ualias2, "user")
+ .join(ualias, Address.user)
+ .join(ualias2, Address.user)
.order_by(Address.id)
.all(),
[
eq_(
sess.query(Address, ualias.concat, ualias.count)
- .join(ualias, "user")
- .join(ualias2, "user")
+ .join(Address.user.of_type(ualias))
+ .join(Address.user.of_type(ualias2))
.order_by(Address.id)
.all(),
[
eq_(
list(
sess.query(Address)
- .join("user")
+ .join(Address.user)
.with_entities(Address.id, User.id, User.concat, User.count)
),
[
eq_(
list(
sess.query(Address, ua)
- .select_entity_from(join(Address, ua, "user"))
+ .select_entity_from(join(Address, ua, Address.user))
.with_entities(Address.id, ua.id, ua.concat, ua.count)
),
[
def go():
o1 = (
sess.query(Order)
- .options(joinedload("address").joinedload("user"))
+ .options(joinedload(Order.address).joinedload(Address.user))
.get(1)
)
eq_(o1.address.user.count, 1)
def go():
o1 = (
sess.query(Order)
- .options(joinedload("address").joinedload("user"))
+ .options(joinedload(Order.address).joinedload(Address.user))
.first()
)
eq_(o1.address.user.count, 1)
"""Query.join"""
User, Address = self.classes.User, self.classes.Address
+ Order = self.classes.Order
session = fixture_session()
q = (
session.query(User)
- .join("orders", "addresses")
+ .outerjoin(User.orders)
+ .outerjoin(Order.addresses)
.filter(Address.id == 1)
)
eq_([User(id=7)], q.all())
session = fixture_session()
q = (
session.query(User)
- .outerjoin("orders", "addresses")
+ .outerjoin(User.orders)
+ .outerjoin(Order.addresses)
.filter(sa.or_(Order.id == None, Address.id == 1))
) # noqa
eq_(set([User(id=7), User(id=8), User(id=10)]), set(q.all()))
q = (
session.query(User)
- .outerjoin("orders", "addresses")
+ .outerjoin(User.orders)
+ .outerjoin(Order.addresses)
.filter(sa.or_(Order.id == None, Address.id == 1))
) # noqa
eq_(q.count(), 4)
[
"relationship",
"relationship_only",
- "string_relationship",
- "string_relationship_only",
"none",
"explicit",
"table_none",
],
[True, False],
)
- ).difference(
- [
- ("string_relationship", False),
- ("string_relationship_only", False),
- ]
),
argnames="onclause_type, use_legacy",
)
if onclause_type == "relationship":
q = q.join(Address, User.addresses)
- elif onclause_type == "string_relationship":
- q = q.join(Address, "addresses")
elif onclause_type == "relationship_only":
q = q.join(User.addresses)
- elif onclause_type == "string_relationship_only":
- q = q.join("addresses")
elif onclause_type == "none":
q = q.join(Address)
elif onclause_type == "explicit":
[Order(id=1), Order(id=5)],
fixture_session()
.query(closed_mapper)
- .with_parent(user, property="closed_orders")
+ .with_parent(user, property=User.closed_orders)
.all(),
)
eq_(
[Order(id=3)],
fixture_session()
.query(open_mapper)
- .with_parent(user, property="open_orders")
+ .with_parent(user, property=User.open_orders)
.all(),
)
eq_(
self.static.item_keyword_result[0:2],
- q.join("keywords").filter(keywords.c.name == "red").all(),
+ q.join(Item.keywords).filter(keywords.c.name == "red").all(),
)
def test_uses_get(self):
session = fixture_session()
q = (
session.query(Item)
- .join("keywords")
+ .join(Item.keywords)
.distinct()
.filter(Keyword.name == "red")
)
h1.h2s.extend([H2("abc"), H2("def")])
s.flush()
- h1s = s.query(H1).options(sa.orm.joinedload("h2s")).all()
+ h1s = s.query(H1).options(sa.orm.joinedload(H1.h2s)).all()
eq_(len(h1s), 5)
self.assert_unordered_result(
{"h2s": (H2, [{"value": "def"}])},
)
- h1s = s.query(H1).options(sa.orm.joinedload("h3s")).all()
+ h1s = s.query(H1).options(sa.orm.joinedload(H1.h3s)).all()
eq_(len(h1s), 5)
h1s = (
s.query(H1)
.options(
- sa.orm.joinedload("t6a").joinedload("h1b"),
- sa.orm.joinedload("h2s"),
- sa.orm.joinedload("h3s").joinedload("h1s"),
+ sa.orm.joinedload(H1.t6a).joinedload(H6.h1b),
+ sa.orm.joinedload(H1.h2s),
+ sa.orm.joinedload(H1.h3s).joinedload(H3.h1s),
)
.all()
)
if use_bound:
stmt = select(User).options(
- Load(User).load_only("name"),
+ Load(User).load_only(User.name),
)
is_true(um.configured)
else:
stmt = select(User).options(
- load_only("name"),
+ load_only(User.name),
)
is_false(um.configured)
sess.commit()
sess2 = fixture_session()
- u2 = sess2.query(User).options(sa.orm.joinedload("addresses")).get(7)
+ u2 = (
+ sess2.query(User).options(sa.orm.joinedload(User.addresses)).get(7)
+ )
sess3 = fixture_session()
u3 = sess3.merge(u2, load=False) # noqa
sess = fixture_session()
eq_(
sess.query(Company)
- .join(Company.employees.of_type(Engineer), "machines")
+ .join(Company.employees.of_type(Engineer))
+ .join(Engineer.machines)
.filter(Machine.name.ilike("%thinkpad%"))
.all(),
[self.c1],
strategy_options._UnboundLoad.joinedload, arg, True, {}
)
- def test_get_path_one_level_string(self):
- User = self.classes.User
-
- sess = fixture_session()
- q = sess.query(User)
-
- opt = self._option_fixture("addresses")
- self._assert_path_result(opt, q, [(User, "addresses")])
-
def test_get_path_one_level_attribute(self):
User = self.classes.User
def test_get_path_one_level_with_unrelated(self):
Order = self.classes.Order
+ User = self.classes.User
sess = fixture_session()
q = sess.query(Order)
- opt = self._option_fixture("addresses")
+ opt = self._option_fixture(User.addresses)
self._assert_path_result(opt, q, [])
- def test_path_multilevel_string(self):
- Item, User, Order = (
- self.classes.Item,
- self.classes.User,
- self.classes.Order,
- )
-
- sess = fixture_session()
- q = sess.query(User)
-
- opt = self._option_fixture("orders.items.keywords")
- self._assert_path_result(
- opt,
- q,
- [
- (User, "orders"),
- (User, "orders", Order, "items"),
- (User, "orders", Order, "items", Item, "keywords"),
- ],
- )
-
def test_path_multilevel_attribute(self):
Item, User, Order = (
self.classes.Item,
],
)
- def test_with_current_matching_string(self):
- Item, User, Order = (
- self.classes.Item,
- self.classes.User,
- self.classes.Order,
- )
-
- sess = fixture_session()
- q = sess.query(Item)._with_current_path(
- self._make_path_registry([User, "orders", Order, "items"])
- )
-
- opt = self._option_fixture("orders.items.keywords")
- self._assert_path_result(opt, q, [(Item, "keywords")])
-
def test_with_current_matching_attribute(self):
Item, User, Order = (
self.classes.Item,
opt = self._option_fixture(User.orders, Order.items, Item.keywords)
self._assert_path_result(opt, q, [(Item, "keywords")])
- def test_with_current_nonmatching_string(self):
- Item, User, Order = (
- self.classes.Item,
- self.classes.User,
- self.classes.Order,
- )
-
- sess = fixture_session()
- q = sess.query(Item)._with_current_path(
- self._make_path_registry([User, "orders", Order, "items"])
- )
-
- opt = self._option_fixture("keywords")
- self._assert_path_result(opt, q, [])
-
- opt = self._option_fixture("items.keywords")
- self._assert_path_result(opt, q, [])
-
def test_with_current_nonmatching_attribute(self):
Item, User, Order = (
self.classes.Item,
q = sess.query(Item, Order)
self._assert_path_result(opt, q, [(Order, "items")])
- def test_multi_entity_opt_on_string(self):
- Item = self.classes.Item
- Order = self.classes.Order
- opt = self._option_fixture("items")
- sess = fixture_session()
- q = sess.query(Item, Order)
- self._assert_path_result(opt, q, [])
-
def test_multi_entity_no_mapped_entities(self):
Item = self.classes.Item
Order = self.classes.Order
Order = self.classes.Order
sess = fixture_session()
q = sess.query(User)
- opt = self._option_fixture(User.orders).joinedload("items")
+ opt = self._option_fixture(User.orders).joinedload(Order.items)
self._assert_path_result(
opt, q, [(User, "orders"), (User, "orders", Order, "items")]
)
- def test_chained_plus_dotted(self):
- User = self.classes.User
- Order = self.classes.Order
- Item = self.classes.Item
- sess = fixture_session()
- q = sess.query(User)
- opt = self._option_fixture("orders.items").joinedload("keywords")
- self._assert_path_result(
- opt,
- q,
- [
- (User, "orders"),
- (User, "orders", Order, "items"),
- (User, "orders", Order, "items", Item, "keywords"),
- ],
- )
-
def test_chained_plus_multi(self):
User = self.classes.User
Order = self.classes.Order
sess = fixture_session()
q = sess.query(User)
opt = self._option_fixture(User.orders, Order.items).joinedload(
- "keywords"
+ Item.keywords
)
self._assert_path_result(
opt,
run_inserts = None
run_deletes = None
- def test_option_with_mapper_basestring(self):
- Item = self.classes.Item
-
- self._assert_option([Item], "keywords")
-
def test_option_with_mapper_PropCompatator(self):
Item = self.classes.Item
self._assert_option([Item], Item.keywords)
- def test_option_with_mapper_then_column_basestring(self):
- Item = self.classes.Item
-
- self._assert_option([Item, Item.id], "keywords")
-
def test_option_with_mapper_then_column_PropComparator(self):
Item = self.classes.Item
self._assert_option([Item, Item.id], Item.keywords)
- def test_option_with_column_then_mapper_basestring(self):
- Item = self.classes.Item
-
- self._assert_option([Item.id, Item], "keywords")
-
def test_option_with_column_then_mapper_PropComparator(self):
Item = self.classes.Item
self._assert_option([Item.id, Item], Item.keywords)
- def test_option_with_column_basestring(self):
- Item = self.classes.Item
-
- message = (
- "Query has only expression-based entities - can't "
- 'find property named "keywords".'
- )
- self._assert_eager_with_just_column_exception(
- Item.id, "keywords", message
- )
-
def test_option_with_column_PropComparator(self):
Item = self.classes.Item
"the root entities to the target attribute. ",
)
- def test_option_against_nonexistent_basestring(self):
- Item = self.classes.Item
- self._assert_eager_with_entity_exception(
- [Item],
- (joinedload("foo"),),
- 'Can\'t find property named "foo" on mapped class '
- "Item->items in this Query.",
- )
-
- def test_option_against_nonexistent_twolevel_basestring(self):
- Item = self.classes.Item
- self._assert_eager_with_entity_exception(
- [Item],
- (joinedload("keywords.foo"),),
- 'Can\'t find property named "foo" on mapped class '
- "Keyword->keywords in this Query.",
- )
-
- def test_option_against_nonexistent_twolevel_chained(self):
- Item = self.classes.Item
- self._assert_eager_with_entity_exception(
- [Item],
- (joinedload("keywords").joinedload("foo"),),
- 'Can\'t find property named "foo" on mapped class '
- "Keyword->keywords in this Query.",
- )
-
- @testing.fails_if(
- lambda: True,
- "PropertyOption doesn't yet check for relation/column on end result",
- )
- def test_option_against_non_relation_basestring(self):
- Item = self.classes.Item
- Keyword = self.classes.Keyword
- self._assert_eager_with_entity_exception(
- [Keyword, Item],
- (joinedload("keywords"),),
- r"Attribute 'keywords' of entity 'Mapper\|Keyword\|keywords' "
- "does not refer to a mapped entity",
- )
-
- @testing.fails_if(
- lambda: True,
- "PropertyOption doesn't yet check for relation/column on end result",
- )
- def test_option_against_multi_non_relation_basestring(self):
- Item = self.classes.Item
- Keyword = self.classes.Keyword
- self._assert_eager_with_entity_exception(
- [Keyword, Item],
- (joinedload("keywords"),),
- r"Attribute 'keywords' of entity 'Mapper\|Keyword\|keywords' "
- "does not refer to a mapped entity",
- )
-
- def test_option_against_wrong_entity_type_basestring(self):
- Item = self.classes.Item
- self._assert_loader_strategy_exception(
- [Item],
- (joinedload("id").joinedload("keywords"),),
- 'Can\'t apply "joined loader" strategy to property "Item.id", '
- 'which is a "column property"; this loader strategy is '
- 'intended to be used with a "relationship property".',
- )
-
- def test_col_option_against_relationship_basestring(self):
- Item = self.classes.Item
- self._assert_loader_strategy_exception(
- [Item],
- (load_only("keywords"),),
- 'Can\'t apply "column loader" strategy to property '
- '"Item.keywords", which is a "relationship property"; this '
- 'loader strategy is intended to be used with a "column property".',
- )
-
def test_load_only_against_multi_entity_attr(self):
User = self.classes.User
Item = self.classes.Item
'loader strategy is intended to be used with a "column property".',
)
- def test_option_against_multi_non_relation_twolevel_basestring(self):
- Item = self.classes.Item
- Keyword = self.classes.Keyword
- self._assert_loader_strategy_exception(
- [Keyword, Item],
- (joinedload("id").joinedload("keywords"),),
- 'Can\'t apply "joined loader" strategy to property "Keyword.id", '
- 'which is a "column property"; this loader strategy is intended '
- 'to be used with a "relationship property".',
- )
-
- def test_option_against_multi_nonexistent_basestring(self):
- Item = self.classes.Item
- Keyword = self.classes.Keyword
- self._assert_eager_with_entity_exception(
- [Keyword, Item],
- (joinedload("description"),),
- 'Can\'t find property named "description" on mapped class '
- "Keyword->keywords in this Query.",
- )
-
- def test_option_against_multi_no_entities_basestring(self):
- Item = self.classes.Item
- Keyword = self.classes.Keyword
- self._assert_eager_with_entity_exception(
- [Keyword.id, Item.id],
- (joinedload("keywords"),),
- r"Query has only expression-based entities - can't find property "
- 'named "keywords".',
- )
-
def test_option_against_wrong_multi_entity_type_attr_one(self):
Item = self.classes.Item
Keyword = self.classes.Keyword
._compile_state,
)
- def test_missing_str_attr_of_type_subclass(self):
- s = fixture_session()
-
- assert_raises_message(
- sa.exc.ArgumentError,
- r'Can\'t find property named "manager_name" on '
- r"mapped class Engineer->engineers in this Query.$",
- s.query(Company)
- .options(
- joinedload(Company.employees.of_type(Engineer)).load_only(
- "manager_name"
- )
- )
- ._compile_state,
- )
-
def test_missing_attr_of_type_wpoly_subclass(self):
s = fixture_session()
sess = fixture_session()
self._assert_opts(sess.query(User), sub_opt, non_sub_opts)
- def test_four_strings(self):
- User, Address, Order, Item, SubItem, Keyword = self.classes(
- "User", "Address", "Order", "Item", "SubItem", "Keyword"
- )
- sub_opt = joinedload("orders").options(
- defer("description"),
- joinedload("items").options(
- joinedload("keywords").options(defer("name")),
- defer("description"),
- ),
- )
- non_sub_opts = [
- joinedload(User.orders),
- defaultload(User.orders).defer(Order.description),
- defaultload(User.orders).joinedload(Order.items),
- defaultload(User.orders)
- .defaultload(Order.items)
- .joinedload(Item.keywords),
- defaultload(User.orders)
- .defaultload(Order.items)
- .defer(Item.description),
- defaultload(User.orders)
- .defaultload(Order.items)
- .defaultload(Item.keywords)
- .defer(Keyword.name),
- ]
- sess = fixture_session()
- self._assert_opts(sess.query(User), sub_opt, non_sub_opts)
-
def test_five(self):
User, Address, Order, Item, SubItem, Keyword = self.classes(
"User", "Address", "Order", "Item", "SubItem", "Keyword"
sess = fixture_session()
self._assert_opts(sess.query(User), sub_opt, non_sub_opts)
- def test_five_strings(self):
- User, Address, Order, Item, SubItem, Keyword = self.classes(
- "User", "Address", "Order", "Item", "SubItem", "Keyword"
- )
- sub_opt = joinedload("orders").options(load_only("description"))
- non_sub_opts = [
- joinedload(User.orders),
- defaultload(User.orders).load_only(Order.description),
- ]
- sess = fixture_session()
- self._assert_opts(sess.query(User), sub_opt, non_sub_opts)
-
def test_invalid_one(self):
User, Address, Order, Item, SubItem = self.classes(
"User", "Address", "Order", "Item", "SubItem"
sess = fixture_session()
self._assert_opts(sess.query(User), sub_opt, non_sub_opts)
- def test_invalid_two(self):
- User, Address, Order, Item, SubItem = self.classes(
- "User", "Address", "Order", "Item", "SubItem"
- )
-
- # these options are "invalid", in that User.orders -> Item.keywords
- # is not a path. However, the "normal" option is not generating
- # an error for now, which is bad, but we're testing here only that
- # it works the same way, so there you go. If and when we make this
- # case raise, then both cases should raise in the same way.
- sub_opt = joinedload("orders").options(
- joinedload("keywords"), joinedload("items")
- )
- non_sub_opts = [
- joinedload(User.orders).joinedload(Item.keywords),
- defaultload(User.orders).joinedload(Order.items),
- ]
- sess = fixture_session()
- self._assert_opts(sess.query(User), sub_opt, non_sub_opts)
-
def test_not_implemented_fromload(self):
User, Address, Order, Item, SubItem = self.classes(
"User", "Address", "Order", "Item", "SubItem"
u = (
sess.query(User)
.order_by(User.id)
- .options(sa.orm.joinedload("adlist"))
+ .options(sa.orm.joinedload(User.adlist))
.filter_by(name="jack")
).one()
eq_(u.adlist, [self.static.user_address_result[0].addresses[0]])
u1 = (
sess.query(User)
.options(
- sa.orm.defer("name"),
- sa.orm.defer("addresses.email_address"),
+ sa.orm.defer(User.name),
+ sa.orm.defaultload(User.addresses).defer(
+ Address.email_address
+ ),
)
.get(u1.id)
)
@testing.requires.non_broken_pickle
@testing.combinations(
lambda User: sa.orm.joinedload(User.addresses),
- lambda: sa.orm.joinedload("addresses"),
- lambda: sa.orm.defer("name"),
lambda User: sa.orm.defer(User.name),
- lambda Address: sa.orm.joinedload("addresses").joinedload(
+ lambda Address: sa.orm.joinedload(User.addresses).joinedload(
Address.dingaling
),
- lambda: sa.orm.joinedload("addresses").raiseload("*"),
+ lambda: sa.orm.joinedload(User.addresses).raiseload("*"),
lambda: sa.orm.raiseload("*"),
)
def test_unbound_options(self, test_case):
@testing.requires.non_broken_pickle
@testing.combinations(
lambda User: sa.orm.Load(User).joinedload(User.addresses),
- lambda User: sa.orm.Load(User).joinedload("addresses"),
- lambda User: sa.orm.Load(User).joinedload("addresses").raiseload("*"),
- lambda User: sa.orm.Load(User).defer("name"),
+ lambda User: sa.orm.Load(User)
+ .joinedload(User.addresses)
+ .raiseload("*"),
lambda User: sa.orm.Load(User).defer(User.name),
lambda User, Address: sa.orm.Load(User)
- .joinedload("addresses")
+ .joinedload(User.addresses)
.joinedload(Address.dingaling),
lambda User, Address: sa.orm.Load(User)
- .joinedload("addresses", innerjoin=True)
+ .joinedload(User.addresses, innerjoin=True)
.joinedload(Address.dingaling),
)
def test_bound_options(self, test_case):
for opt in [
sa.orm.joinedload(User.addresses),
- sa.orm.joinedload("addresses"),
- sa.orm.defer("name"),
sa.orm.defer(User.name),
- sa.orm.joinedload("addresses").joinedload(Address.dingaling),
+ sa.orm.joinedload(User.addresses).joinedload(Address.dingaling),
]:
context = sess.query(User).options(opt)._compile_context()
opt = [
s = fixture_session()
- q = s.query(User).join("addresses").filter(Address.user_id == 8)
+ q = s.query(User).join(User.addresses).filter(Address.user_id == 8)
assert_raises(sa_exc.InvalidRequestError, q.get, 7)
assert_raises(
sa_exc.InvalidRequestError,
s.query(User).get(7)
- q = s.query(User).join("addresses").filter(Address.user_id == 8)
+ q = s.query(User).join(User.addresses).filter(Address.user_id == 8)
assert_raises(sa_exc.InvalidRequestError, q.get, 7)
def test_unique_param_names(self):
def test_populate_existing(self):
User, Address = self.classes.User, self.classes.Address
+ Order = self.classes.Order
s = fixture_session(autoflush=False)
# eager load does
s.query(User).options(
- joinedload("addresses"), joinedload("orders").joinedload("items")
+ joinedload(User.addresses),
+ joinedload(User.orders).joinedload(Order.items),
).populate_existing().all()
assert u.addresses[0].email_address == "jack@bean.com"
assert u.orders[1].items[2].description == "item 5"
def test_populate_existing_future(self):
User, Address = self.classes.User, self.classes.Address
+ Order = self.classes.Order
s = fixture_session(autoflush=False)
stmt = (
select(User)
.options(
- joinedload("addresses"),
- joinedload("orders").joinedload("items"),
+ joinedload(User.addresses),
+ joinedload(User.orders).joinedload(Order.items),
)
.execution_options(populate_existing=True)
)
q = testing.resolve_lambda(test_case, User=User, s=s)
- assert_raises(sa_exc.InvalidRequestError, q.join, "addresses")
+ assert_raises(sa_exc.InvalidRequestError, q.join, User.addresses)
assert_raises(sa_exc.InvalidRequestError, q.filter, User.name == "ed")
assert_raises(sa_exc.InvalidRequestError, q.having, "foo")
- q.enable_assertions(False).join("addresses")
+ q.enable_assertions(False).join(User.addresses)
q.enable_assertions(False).filter(User.name == "ed")
q.enable_assertions(False).order_by("foo")
q.enable_assertions(False).group_by("foo")
q = s.query(User).select_from(users)
assert_raises(sa_exc.InvalidRequestError, q.select_from, users)
- q = s.query(User).join("addresses")
+ q = s.query(User).join(User.addresses)
assert_raises(sa_exc.InvalidRequestError, q.select_from, users)
q = s.query(User).order_by(User.id)
# test that the contents are not adapted by the aliased join
ua = aliased(Address)
assert [User(id=7), User(id=8)] == sess.query(User).join(
- ua, "addresses"
+ ua, User.addresses
).filter(
~User.addresses.any(Address.email_address == "fred@fred.com")
).all()
assert [User(id=10)] == sess.query(User).outerjoin(
- ua, "addresses"
+ ua, User.addresses
).filter(~User.addresses.any()).all()
def test_any_doesnt_overcorrelate(self):
# test that any() doesn't overcorrelate
assert [User(id=7), User(id=8)] == sess.query(User).join(
- "addresses"
+ User.addresses
).filter(
~User.addresses.any(Address.email_address == "fred@fred.com")
).all()
# test has() doesn't overcorrelate
assert [Address(id=2), Address(id=3), Address(id=4)] == sess.query(
Address
- ).join("user").filter(
+ ).join(Address.user).filter(
Address.user.has(User.name.like("%ed%"), id=8)
).order_by(
Address.id
ua = aliased(User)
assert [Address(id=2), Address(id=3), Address(id=4)] == sess.query(
Address
- ).join(ua, "user").filter(
+ ).join(ua, Address.user).filter(
Address.user.has(User.name.like("%ed%"), id=8)
).order_by(
Address.id
sess = fixture_session()
assert [User(name="ed", id=8)] == sess.query(User).order_by(
User.id
- ).group_by(User).join("addresses").having(
+ ).group_by(User).join(User.addresses).having(
func.count(Address.id) > 2
).all()
User(name="jack", id=7),
User(name="fred", id=9),
] == sess.query(User).order_by(User.id).group_by(User).join(
- "addresses"
+ User.addresses
).having(
func.count(Address.id) < 2
).all()
subq = (
sess.query(User, Address.email_address)
- .join("addresses")
+ .join(User.addresses)
.distinct()
.subquery()
)
sess = fixture_session()
q = (
sess.query(User, Address.email_address)
- .join("addresses")
+ .join(User.addresses)
.distinct()
)
# test that it works on embedded joinedload/LIMIT subquery
q = (
sess.query(User)
- .join("addresses")
+ .join(User.addresses)
.distinct()
- .options(joinedload("addresses"))
+ .options(joinedload(User.addresses))
.order_by(desc(Address.email_address))
.limit(2)
)
q = (
sess.query(User)
- .join("addresses")
+ .join(User.addresses)
.distinct(Address.email_address)
- .options(joinedload("addresses"))
+ .options(joinedload(User.addresses))
.order_by(desc(Address.email_address))
.limit(2)
)
User = self.classes.User
sess = fixture_session()
- q = sess.query(User).options(joinedload("addresses")).yield_per(1)
+ q = sess.query(User).options(joinedload(User.addresses)).yield_per(1)
assert_raises_message(
sa_exc.InvalidRequestError,
"Can't use yield_per with eager loaders that require "
User = self.classes.User
sess = fixture_session()
- q = sess.query(User).options(subqueryload("addresses")).yield_per(1)
+ q = sess.query(User).options(subqueryload(User.addresses)).yield_per(1)
assert_raises_message(
sa_exc.InvalidRequestError,
"Can't use yield_per with eager loaders that require "
sess = fixture_session()
q = (
sess.query(User)
- .options(subqueryload("addresses"))
+ .options(subqueryload(User.addresses))
.enable_eagerloads(False)
.yield_per(1)
)
q = (
sess.query(User)
- .options(joinedload("addresses"))
+ .options(joinedload(User.addresses))
.enable_eagerloads(False)
.yield_per(1)
)
sess = fixture_session()
q = (
sess.query(Address)
- .options(lazyload("*"), joinedload("user"))
+ .options(lazyload("*"), joinedload(Address.user))
.yield_per(1)
.filter_by(id=1)
)
q = (
s.query(User)
- .options(joinedload("addresses"))
+ .options(joinedload(User.addresses))
.order_by(desc("name"))
.limit(1)
)
q = (
s.query(User)
- .options(joinedload("addresses"))
+ .options(joinedload(User.addresses))
.order_by("name")
.limit(1)
)
self.assert_compile(
s.query(User)
- .options(joinedload("addresses"))
+ .options(joinedload(User.addresses))
.order_by("users_name")
.limit(1),
"SELECT anon_1.users_id AS anon_1_users_id, "
# however! this works (again?)
eq_(
s.query(User)
- .options(joinedload("addresses"))
+ .options(joinedload(User.addresses))
.order_by("users_name")
.first(),
User(name="chuck", addresses=[]),
self.assert_compile(
s.query(User)
- .options(joinedload("addresses"))
+ .options(joinedload(User.addresses))
.order_by(desc("users_name"))
.limit(1),
"SELECT anon_1.users_id AS anon_1_users_id, "
# however! this works (again?)
eq_(
s.query(User)
- .options(joinedload("addresses"))
+ .options(joinedload(User.addresses))
.order_by(desc("users_name"))
.first(),
User(name="jack", addresses=[Address()]),
q = sess.query(User, Address.email_address.label("email_address"))
result = (
- q.join("addresses")
+ q.join(User.addresses)
.options(joinedload(User.orders))
.order_by("email_address desc")
.limit(1)
] == o
# test with explicit property
- o = sess.query(Order).with_parent(u1, property="orders").all()
- assert [
- Order(description="order 1"),
- Order(description="order 3"),
- Order(description="order 5"),
- ] == o
-
o = sess.query(Order).with_parent(u1, property=User.orders).all()
assert [
Order(description="order 1"),
sess = fixture_session()
u1 = sess.query(User).get(7)
q = sess.query(User, Address).filter(
- with_parent(u1, "addresses", from_entity=Address)
+ with_parent(u1, User.addresses, from_entity=Address)
)
self.assert_compile(
q,
sess = fixture_session()
u1 = sess.query(User).get(7)
q = sess.query(User, Address).with_parent(
- u1, "addresses", from_entity=Address
+ u1, User.addresses, from_entity=Address
)
self.assert_compile(
q,
sess = fixture_session()
u1 = sess.query(User).get(7)
a1 = aliased(Address)
- q = sess.query(a1).with_parent(u1, "addresses")
+ q = sess.query(a1).with_parent(u1, User.addresses)
self.assert_compile(
q,
"SELECT addresses_1.id AS addresses_1_id, "
q = sess.query(User)
u1 = q.filter_by(name="jack").one()
utrans = User(id=u1.id)
- o = sess.query(Order).with_parent(utrans, "orders")
+ o = sess.query(Order).with_parent(utrans, User.orders)
eq_(
[
Order(description="order 1"),
o.all(),
)
- o = sess.query(Order).filter(with_parent(utrans, "orders"))
+ o = sess.query(Order).filter(with_parent(utrans, User.orders))
eq_(
[
Order(description="order 1"),
opending = Order(id=20, user_id=o1.user_id)
sess.add(opending)
eq_(
- sess.query(User).with_parent(opending, "user").one(),
+ sess.query(User).with_parent(opending, Order.user).one(),
User(id=o1.user_id),
)
eq_(
- sess.query(User).filter(with_parent(opending, "user")).one(),
+ sess.query(User).filter(with_parent(opending, Order.user)).one(),
User(id=o1.user_id),
)
opending = Order(user_id=o1.user_id)
sess.add(opending)
eq_(
- sess.query(User).with_parent(opending, "user").one(),
+ sess.query(User).with_parent(opending, Order.user).one(),
User(id=o1.user_id),
)
sess = fixture_session()
u1, u2 = sess.query(User).order_by(User.id)[0:2]
- q1 = sess.query(Address).with_parent(u1, "addresses")
- q2 = sess.query(Address).with_parent(u2, "addresses")
+ q1 = sess.query(Address).with_parent(u1, User.addresses)
+ q2 = sess.query(Address).with_parent(u2, User.addresses)
self.assert_compile(
q1.union(q2),
self.assert_compile(
sess.query(Address).filter(
- or_(with_parent(u1, "addresses"), with_parent(u2, "addresses"))
+ or_(
+ with_parent(u1, User.addresses),
+ with_parent(u2, User.addresses),
+ )
),
"SELECT addresses.id AS addresses_id, addresses.user_id AS "
"addresses_user_id, addresses.email_address AS "
sess = fixture_session()
- q = sess.query(User).with_parent(Address(user_id=None), "user")
+ q = sess.query(User).with_parent(Address(user_id=None), Address.user)
with expect_warnings("Got None for value of column"):
self.assert_compile(
q,
s = fixture_session()
q = s.query(User).with_parent(
- Address(user_id=None, email_address=None), "special_user"
+ Address(user_id=None, email_address=None), Address.special_user
)
with expect_warnings("Got None for value of column"):
self.assert_sql_count(testing.db, go, 1)
- def test_options_syn_of_syn_string(self):
- User, Order = self.classes.User, self.classes.Order
-
- s = fixture_session()
-
- def go():
- result = (
- s.query(User)
- .filter_by(name="jack")
- .options(joinedload("orders_syn_2"))
- .all()
- )
- eq_(
- result,
- [
- User(
- id=7,
- name="jack",
- orders=[
- Order(description="order 1"),
- Order(description="order 3"),
- Order(description="order 5"),
- ],
- )
- ],
- )
-
- self.assert_sql_count(testing.db, go, 1)
-
def test_joins(self):
User, Order = self.classes.User, self.classes.Order
Order, User = self.classes.Order, self.classes.User
for nameprop, orderprop in (
- ("name", "orders"),
- ("name_syn", "orders"),
- ("name", "orders_syn"),
- ("name", "orders_syn_2"),
- ("name_syn", "orders_syn"),
- ("name_syn", "orders_syn_2"),
+ ("name", User.orders),
+ ("name_syn", User.orders),
+ ("name", User.orders_syn),
+ ("name", User.orders_syn_2),
+ ("name_syn", User.orders_syn),
+ ("name_syn", User.orders_syn_2),
):
with fixture_session() as sess:
q = sess.query(User)
sess.commit()
# test that was broken by #3060
- a1 = sess.query(A).options(joinedload("b")).first()
+ a1 = sess.query(A).options(joinedload(A.b)).first()
a1.bid = b1.id
sess.flush()
users,
properties=dict(addresses=relationship(Address, lazy="raise")),
)
- q = fixture_session().query(User).options(sa.orm.lazyload("addresses"))
+ q = (
+ fixture_session()
+ .query(User)
+ .options(sa.orm.lazyload(User.addresses))
+ )
result = [None]
def go():
a1 = (
s.query(Address)
.filter_by(id=1)
- .options(sa.orm.raiseload("user"))
+ .options(sa.orm.raiseload(Address.user))
.first()
)
a1 = (
s.query(Address)
.filter_by(id=1)
- .options(sa.orm.raiseload("user", sql_only=True))
+ .options(sa.orm.raiseload(Address.user, sql_only=True))
.first()
)
a1 = (
s.query(Address)
.filter_by(id=1)
- .options(sa.orm.raiseload("user", sql_only=True))
+ .options(sa.orm.raiseload(Address.user, sql_only=True))
.first()
)
assert "user" not in a1.__dict__
a1 = (
s.query(Address)
.filter_by(id=1)
- .options(sa.orm.raiseload("user", sql_only=True))
+ .options(sa.orm.raiseload(Address.user, sql_only=True))
.first()
)
def go():
eq_(
self.static.item_keyword_result[0:2],
- q.join("keywords").filter(Keyword.name == "red").all(),
+ q.join(Item.keywords).filter(Keyword.name == "red").all(),
)
self.assert_sql_count(testing.db, go, 2)
ka = aliased(Keyword)
eq_(
self.static.item_keyword_result[0:2],
- (q.join(ka, "keywords").filter(ka.name == "red")).all(),
+ (q.join(ka, Item.keywords).filter(ka.name == "red")).all(),
)
self.assert_sql_count(testing.db, go, 2)
a2 = u1.addresses[0]
a2.email_address = "foo"
sess.query(User).options(
- selectinload("addresses").selectinload("dingaling")
+ selectinload(User.addresses).selectinload(Address.dingaling)
).filter_by(id=8).all()
assert u1.addresses[-1] is a1
for a in u1.addresses:
o1 = Order()
u1.orders.append(o1)
sess.query(User).options(
- selectinload("orders").selectinload("items")
+ selectinload(User.orders).selectinload(Order.items)
).filter_by(id=7).all()
for o in u1.orders:
if o is not o1:
u1 = (
sess.query(User)
.filter_by(id=8)
- .options(selectinload("addresses"))
+ .options(selectinload(User.addresses))
.one()
)
sess.query(User).filter_by(id=8).options(
- selectinload("addresses").selectinload("dingaling")
+ selectinload(User.addresses).selectinload(Address.dingaling)
).first()
assert "dingaling" in u1.addresses[0].__dict__
u1 = (
sess.query(User)
.filter_by(id=7)
- .options(selectinload("orders"))
+ .options(selectinload(User.orders))
.one()
)
sess.query(User).filter_by(id=7).options(
- selectinload("orders").selectinload("items")
+ selectinload(User.orders).selectinload(Order.items)
).first()
assert "items" in u1.orders[0].__dict__
eq_(
Node(data="n1", children=[Node(data="n11"), Node(data="n12")]),
sess.query(Node)
- .options(undefer("data"))
+ .options(undefer(Node.data))
.order_by(Node.id)
.first(),
)
eq_(
Node(data="n1", children=[Node(data="n11"), Node(data="n12")]),
sess.query(Node)
- .options(undefer("data"), undefer("children.data"))
+ .options(
+ undefer(Node.data),
+ defaultload(Node.children).undefer(Node.data),
+ )
.first(),
)
sess.query(Node)
.filter_by(data="n1")
.order_by(Node.id)
- .options(selectinload("children").selectinload("children"))
+ .options(
+ selectinload(Node.children).selectinload(Node.children)
+ )
.first()
)
eq_(
from sqlalchemy.orm import backref
from sqlalchemy.orm import clear_mappers
from sqlalchemy.orm import close_all_sessions
+from sqlalchemy.orm import defaultload
from sqlalchemy.orm import defer
from sqlalchemy.orm import deferred
from sqlalchemy.orm import joinedload
def go():
eq_(
self.static.item_keyword_result[0:2],
- q.join("keywords").filter(Keyword.name == "red").all(),
+ q.join(Item.keywords).filter(Keyword.name == "red").all(),
)
self.assert_sql_count(testing.db, go, 2)
ka = aliased(Keyword)
eq_(
self.static.item_keyword_result[0:2],
- (q.join(ka, "keywords").filter(ka.name == "red")).all(),
+ (q.join(ka, Item.keywords).filter(ka.name == "red")).all(),
)
self.assert_sql_count(testing.db, go, 2)
a2 = u1.addresses[0]
a2.email_address = "foo"
sess.query(User).options(
- subqueryload("addresses").subqueryload("dingaling")
+ subqueryload(User.addresses).subqueryload(Address.dingaling)
).filter_by(id=8).all()
assert u1.addresses[-1] is a1
for a in u1.addresses:
o1 = Order()
u1.orders.append(o1)
sess.query(User).options(
- subqueryload("orders").subqueryload("items")
+ subqueryload(User.orders).subqueryload(Order.items)
).filter_by(id=7).all()
for o in u1.orders:
if o is not o1:
u1 = (
sess.query(User)
.filter_by(id=8)
- .options(subqueryload("addresses"))
+ .options(subqueryload(User.addresses))
.one()
)
sess.query(User).filter_by(id=8).options(
- subqueryload("addresses").subqueryload("dingaling")
+ subqueryload(User.addresses).subqueryload(Address.dingaling)
).first()
assert "dingaling" in u1.addresses[0].__dict__
u1 = (
sess.query(User)
.filter_by(id=7)
- .options(subqueryload("orders"))
+ .options(subqueryload(User.orders))
.one()
)
sess.query(User).filter_by(id=7).options(
- subqueryload("orders").subqueryload("items")
+ subqueryload(User.orders).subqueryload(Order.items)
).first()
assert "items" in u1.orders[0].__dict__
eq_(
Node(data="n1", children=[Node(data="n11"), Node(data="n12")]),
sess.query(Node)
- .options(undefer("data"))
+ .options(undefer(Node.data))
.order_by(Node.id)
.first(),
)
eq_(
Node(data="n1", children=[Node(data="n11"), Node(data="n12")]),
sess.query(Node)
- .options(undefer("data"), undefer("children.data"))
+ .options(
+ undefer(Node.data),
+ defaultload(Node.children).undefer(Node.data),
+ )
.first(),
)
sess.query(Node)
.filter_by(data="n1")
.order_by(Node.id)
- .options(subqueryload("children").subqueryload("children"))
+ .options(
+ subqueryload(Node.children).subqueryload(Node.children)
+ )
.first()
)
eq_(
# Parent->subqueryload->Child->joinedload->parent->noload->children.
# the actual subqueryload has to emit *after* we've started populating
# Parent->subqueryload->child.
- parent = s.query(Parent).options([subqueryload("children")]).first()
+ parent = (
+ s.query(Parent).options([subqueryload(Parent.children)]).first()
+ )
eq_(parent.children, [Child(name="c1")])
new_a1 = (
session.query(A)
- .options(sa.orm.joinedload("t2s"))
+ .options(sa.orm.joinedload(A.t2s))
.filter(t1.c.a == a1.a)
).one()
assert new_a1.a == a1.a
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
