"Indicates the version that started raising this deprecation warning"
-class RemovedIn20Warning(SADeprecationWarning):
- """Issued for usage of APIs specifically deprecated in SQLAlchemy 2.0.
+class Base20DeprecationWarning(SADeprecationWarning):
+ """Issued for usage of APIs specifically deprecated or legacy in
+ SQLAlchemy 2.0.
.. seealso::
def __str__(self):
return (
- super(RemovedIn20Warning, self).__str__()
+ super(Base20DeprecationWarning, self).__str__()
+ " (Background on SQLAlchemy 2.0 at: https://sqlalche.me/e/b8d9)"
)
+class LegacyAPIWarning(Base20DeprecationWarning):
+ """indicates an API that is in 'legacy' status, a long term deprecation."""
+
+
+class RemovedIn20Warning(Base20DeprecationWarning):
+ """indicates an API that will be fully removed in SQLAlchemy 2.0."""
+
+
class MovedIn20Warning(RemovedIn20Warning):
"""Subtype of RemovedIn20Warning to indicate an API that moved only."""
we will attempt to derive an expression from based on string name.
"""
+
if self._legacy_setup_joins:
_last_joined_entity = self._last_joined_entity
if _last_joined_entity is not None:
def expect_deprecated_20(*messages, **kw):
- return _expect_warnings(sa_exc.RemovedIn20Warning, messages, **kw)
+ return _expect_warnings(sa_exc.Base20DeprecationWarning, messages, **kw)
def emits_warning_on(db, *messages):
# ORM Query
#
r"The Query\.get\(\) method",
- r"The Query\.with_parent\(\) method",
- r"The Query\.with_parent\(\) method",
- r"The ``aliased`` and ``from_joinpoint`` keyword arguments",
r"The Query.with_polymorphic\(\) method is considered "
"legacy as of the 1.x series",
#
if alternative:
message += " " + alternative
+ if becomes_legacy:
+ warning_cls = exc.LegacyAPIWarning
+ else:
+ warning_cls = exc.RemovedIn20Warning
+
def decorate(cls):
return _decorate_cls_with_warning(
cls,
constructor,
- exc.RemovedIn20Warning,
+ warning_cls,
message,
- exc.RemovedIn20Warning.deprecated_since,
+ warning_cls.deprecated_since,
message,
)
(
[
sa_exceptions.SADeprecationWarning,
+ sa_exceptions.Base20DeprecationWarning,
+ sa_exceptions.LegacyAPIWarning,
sa_exceptions.RemovedIn20Warning,
sa_exceptions.MovedIn20Warning,
sa_exceptions.SAWarning,
--- /dev/null
+from sqlalchemy import ForeignKey
+from sqlalchemy import Integer
+from sqlalchemy import String
+from sqlalchemy import testing
+from sqlalchemy.orm import relationship
+from sqlalchemy.testing import eq_
+from sqlalchemy.testing import fixtures
+from sqlalchemy.testing.assertions import expect_deprecated_20
+from sqlalchemy.testing.fixtures import fixture_session
+from sqlalchemy.testing.schema import Column
+from sqlalchemy.testing.schema import Table
+from ._poly_fixtures import _Polymorphic
+from ._poly_fixtures import _PolymorphicAliasedJoins
+from ._poly_fixtures import _PolymorphicJoins
+from ._poly_fixtures import _PolymorphicPolymorphic
+from ._poly_fixtures import _PolymorphicUnions
+from ._poly_fixtures import Company
+from ._poly_fixtures import Engineer
+from ._poly_fixtures import Manager
+from ._poly_fixtures import Paperwork
+from ._poly_fixtures import Person
+
+
+aliased_jp_dep = (
+ r"The ``aliased`` and ``from_joinpoint`` keyword arguments "
+ r"to Query.join\(\) are deprecated"
+)
+
+
+class _PolymorphicTestBase(fixtures.NoCache):
+ __backend__ = True
+ __dialect__ = "default_enhanced"
+
+ @classmethod
+ def setup_mappers(cls):
+ super(_PolymorphicTestBase, cls).setup_mappers()
+ global people, engineers, managers, boss
+ global companies, paperwork, machines
+ people, engineers, managers, boss, companies, paperwork, machines = (
+ cls.tables.people,
+ cls.tables.engineers,
+ cls.tables.managers,
+ cls.tables.boss,
+ cls.tables.companies,
+ cls.tables.paperwork,
+ cls.tables.machines,
+ )
+
+ @classmethod
+ def insert_data(cls, connection):
+ super(_PolymorphicTestBase, cls).insert_data(connection)
+
+ global all_employees, c1_employees, c2_employees
+ global c1, c2, e1, e2, e3, b1, m1
+ c1, c2, all_employees, c1_employees, c2_employees = (
+ cls.c1,
+ cls.c2,
+ cls.all_employees,
+ cls.c1_employees,
+ cls.c2_employees,
+ )
+ e1, e2, e3, b1, m1 = cls.e1, cls.e2, cls.e3, cls.b1, cls.m1
+
+ def test_join_from_polymorphic_flag_aliased_one(self):
+ sess = fixture_session()
+ with expect_deprecated_20(aliased_jp_dep):
+ eq_(
+ sess.query(Person)
+ .order_by(Person.person_id)
+ .join(Person.paperwork, aliased=True)
+ .filter(Paperwork.description.like("%review%"))
+ .all(),
+ [b1, m1],
+ )
+
+ def test_join_from_polymorphic_flag_aliased_two(self):
+ sess = fixture_session()
+ with expect_deprecated_20(aliased_jp_dep):
+ eq_(
+ sess.query(Person)
+ .order_by(Person.person_id)
+ .join(Person.paperwork, aliased=True)
+ .filter(Paperwork.description.like("%#2%"))
+ .all(),
+ [e1, m1],
+ )
+
+ def test_join_from_with_polymorphic_flag_aliased_one(self):
+ sess = fixture_session()
+ with expect_deprecated_20(aliased_jp_dep):
+ eq_(
+ sess.query(Person)
+ .with_polymorphic(Manager)
+ .join(Person.paperwork, aliased=True)
+ .filter(Paperwork.description.like("%review%"))
+ .all(),
+ [b1, m1],
+ )
+
+ def test_join_from_with_polymorphic_flag_aliased_two(self):
+ sess = fixture_session()
+ with expect_deprecated_20(aliased_jp_dep):
+ eq_(
+ sess.query(Person)
+ .with_polymorphic([Manager, Engineer])
+ .order_by(Person.person_id)
+ .join(Person.paperwork, aliased=True)
+ .filter(Paperwork.description.like("%#2%"))
+ .all(),
+ [e1, m1],
+ )
+
+ def test_join_to_polymorphic_flag_aliased(self):
+ sess = fixture_session()
+ with expect_deprecated_20(aliased_jp_dep):
+ eq_(
+ sess.query(Company)
+ .join(Company.employees, aliased=True)
+ .filter(Person.name == "vlad")
+ .one(),
+ c2,
+ )
+
+ def test_polymorphic_any_flag_alias_two(self):
+ sess = fixture_session()
+ # test that the aliasing on "Person" does not bleed into the
+ # EXISTS clause generated by any()
+ any_ = Company.employees.any(Person.name == "wally")
+ with expect_deprecated_20(aliased_jp_dep):
+ eq_(
+ sess.query(Company)
+ .join(Company.employees, aliased=True)
+ .filter(Person.name == "dilbert")
+ .filter(any_)
+ .all(),
+ [c1],
+ )
+
+ def test_join_from_polymorphic_flag_aliased_three(self):
+ sess = fixture_session()
+ with expect_deprecated_20(aliased_jp_dep):
+ eq_(
+ sess.query(Engineer)
+ .order_by(Person.person_id)
+ .join(Person.paperwork, aliased=True)
+ .filter(Paperwork.description.like("%#2%"))
+ .all(),
+ [e1],
+ )
+
+
+class PolymorphicTest(_PolymorphicTestBase, _Polymorphic):
+ pass
+
+
+class PolymorphicPolymorphicTest(
+ _PolymorphicTestBase, _PolymorphicPolymorphic
+):
+ pass
+
+
+class PolymorphicUnionsTest(_PolymorphicTestBase, _PolymorphicUnions):
+ pass
+
+
+class PolymorphicAliasedJoinsTest(
+ _PolymorphicTestBase, _PolymorphicAliasedJoins
+):
+ pass
+
+
+class PolymorphicJoinsTest(_PolymorphicTestBase, _PolymorphicJoins):
+ pass
+
+
+class RelationshipToSingleTest(
+ testing.AssertsCompiledSQL, fixtures.MappedTest
+):
+ __dialect__ = "default"
+
+ @classmethod
+ def define_tables(cls, metadata):
+ Table(
+ "employees",
+ metadata,
+ Column(
+ "employee_id",
+ Integer,
+ primary_key=True,
+ test_needs_autoincrement=True,
+ ),
+ Column("name", String(50)),
+ Column("manager_data", String(50)),
+ Column("engineer_info", String(50)),
+ Column("type", String(20)),
+ Column("company_id", Integer, ForeignKey("companies.company_id")),
+ )
+
+ Table(
+ "companies",
+ metadata,
+ Column(
+ "company_id",
+ Integer,
+ primary_key=True,
+ test_needs_autoincrement=True,
+ ),
+ Column("name", String(50)),
+ )
+
+ @classmethod
+ def setup_classes(cls):
+ class Company(cls.Comparable):
+ pass
+
+ class Employee(cls.Comparable):
+ pass
+
+ class Manager(Employee):
+ pass
+
+ class Engineer(Employee):
+ pass
+
+ class JuniorEngineer(Engineer):
+ pass
+
+ def test_of_type_aliased_fromjoinpoint(self):
+ Company, Employee, Engineer = (
+ self.classes.Company,
+ self.classes.Employee,
+ self.classes.Engineer,
+ )
+ companies, employees = self.tables.companies, self.tables.employees
+
+ self.mapper_registry.map_imperatively(
+ Company, companies, properties={"employee": relationship(Employee)}
+ )
+ self.mapper_registry.map_imperatively(
+ Employee, employees, polymorphic_on=employees.c.type
+ )
+ self.mapper_registry.map_imperatively(
+ Engineer, inherits=Employee, polymorphic_identity="engineer"
+ )
+
+ sess = fixture_session()
+
+ with expect_deprecated_20(aliased_jp_dep):
+ self.assert_compile(
+ sess.query(Company).outerjoin(
+ Company.employee.of_type(Engineer),
+ aliased=True,
+ from_joinpoint=True,
+ ),
+ "SELECT companies.company_id AS companies_company_id, "
+ "companies.name AS companies_name FROM companies "
+ "LEFT OUTER JOIN employees AS employees_1 ON "
+ "companies.company_id = employees_1.company_id "
+ "AND employees_1.type IN ([POSTCOMPILE_type_1])",
+ )
[m1],
)
- def test_join_from_polymorphic_flag_aliased_one(self):
- sess = fixture_session()
- eq_(
- sess.query(Person)
- .order_by(Person.person_id)
- .join(Person.paperwork, aliased=True)
- .filter(Paperwork.description.like("%review%"))
- .all(),
- [b1, m1],
- )
-
- def test_join_from_polymorphic_flag_aliased_one_future(self):
+ def test_join_from_polymorphic_aliased_one_future(self):
sess = fixture_session(future=True)
pa = aliased(Paperwork)
[b1, m1],
)
- def test_join_from_polymorphic_flag_aliased_two(self):
- sess = fixture_session()
- eq_(
- sess.query(Person)
- .order_by(Person.person_id)
- .join(Person.paperwork, aliased=True)
- .filter(Paperwork.description.like("%#2%"))
- .all(),
- [e1, m1],
- )
-
def test_join_from_polymorphic_explicit_aliased_two(self):
sess = fixture_session()
pa = aliased(Paperwork)
[e1, m1],
)
- def test_join_from_polymorphic_flag_aliased_three(self):
- sess = fixture_session()
- eq_(
- sess.query(Engineer)
- .order_by(Person.person_id)
- .join(Person.paperwork, aliased=True)
- .filter(Paperwork.description.like("%#2%"))
- .all(),
- [e1],
- )
-
def test_join_from_polymorphic_explicit_aliased_three(self):
sess = fixture_session()
pa = aliased(Paperwork)
[m1],
)
- def test_join_from_with_polymorphic_flag_aliased_one(self):
- sess = fixture_session()
- eq_(
- sess.query(Person)
- .with_polymorphic(Manager)
- .join(Person.paperwork, aliased=True)
- .filter(Paperwork.description.like("%review%"))
- .all(),
- [b1, m1],
- )
-
def test_join_from_with_polymorphic_explicit_aliased_one(self):
sess = fixture_session()
pa = aliased(Paperwork)
[b1, m1],
)
- def test_join_from_with_polymorphic_flag_aliased_two(self):
- sess = fixture_session()
- eq_(
- sess.query(Person)
- .with_polymorphic([Manager, Engineer])
- .order_by(Person.person_id)
- .join(Person.paperwork, aliased=True)
- .filter(Paperwork.description.like("%#2%"))
- .all(),
- [e1, m1],
- )
-
def test_join_from_with_polymorphic_explicit_aliased_two(self):
sess = fixture_session()
pa = aliased(Paperwork)
c2,
)
- def test_join_to_polymorphic_flag_aliased(self):
- sess = fixture_session()
- eq_(
- sess.query(Company)
- .join(Company.employees, aliased=True)
- .filter(Person.name == "vlad")
- .one(),
- c2,
- )
-
def test_join_to_polymorphic_explicit_aliased(self):
sess = fixture_session()
ea = aliased(Person)
any_ = Company.employees.any(Person.name == "vlad")
eq_(sess.query(Company).filter(any_).all(), [c2])
- def test_polymorphic_any_flag_alias_two(self):
- sess = fixture_session()
- # test that the aliasing on "Person" does not bleed into the
- # EXISTS clause generated by any()
- any_ = Company.employees.any(Person.name == "wally")
- eq_(
- sess.query(Company)
- .join(Company.employees, aliased=True)
- .filter(Person.name == "dilbert")
- .filter(any_)
- .all(),
- [c1],
- )
-
def test_polymorphic_any_explicit_alias_two(self):
sess = fixture_session()
# test that the aliasing on "Person" does not bleed into the
[Company(name="c1")],
)
- def test_of_type_aliased_fromjoinpoint(self):
- Company, Employee, Engineer = (
- self.classes.Company,
- self.classes.Employee,
- self.classes.Engineer,
- )
- companies, employees = self.tables.companies, self.tables.employees
-
- self.mapper_registry.map_imperatively(
- Company, companies, properties={"employee": relationship(Employee)}
- )
- self.mapper_registry.map_imperatively(
- Employee, employees, polymorphic_on=employees.c.type
- )
- self.mapper_registry.map_imperatively(
- Engineer, inherits=Employee, polymorphic_identity="engineer"
- )
-
- sess = fixture_session()
- self.assert_compile(
- sess.query(Company).outerjoin(
- Company.employee.of_type(Engineer),
- aliased=True,
- from_joinpoint=True,
- ),
- "SELECT companies.company_id AS companies_company_id, "
- "companies.name AS companies_name FROM companies "
- "LEFT OUTER JOIN employees AS employees_1 ON "
- "companies.company_id = employees_1.company_id "
- "AND employees_1.type IN ([POSTCOMPILE_type_1])",
- )
-
def test_join_explicit_onclause_no_discriminator(self):
# test issue #3462
Company, Employee, Engineer = (
from sqlalchemy.orm import relationship
from sqlalchemy.orm import Session
from sqlalchemy.orm import util as orm_util
+from sqlalchemy.orm import with_parent
from sqlalchemy.orm.attributes import instance_state
from sqlalchemy.orm.collections import attribute_mapped_collection
from sqlalchemy.orm.decl_api import declarative_base
assert p1 not in sess
sess.flush()
eq_(
- sess.query(Pref).with_parent(someuser).all(),
+ sess.query(Pref).filter(with_parent(someuser, User.pref)).all(),
[Pref(data="someotherpref")],
)
r"in the ORM with_parent\(\) function"
)
+query_wparent_dep = (
+ r"The Query.with_parent\(\) method is considered legacy as of the 1.x"
+)
+
sef_dep = (
r"The Query.select_entity_from\(\) method is considered "
"legacy as of the 1.x"
assert q.count() == 1
assert [User(id=7)] == q.all()
+ def test_does_filter_aliasing_work(self):
+ User, Address = self.classes("User", "Address")
+
+ s = fixture_session()
+
+ # aliased=True is to be deprecated, other filter lambdas
+ # that go into effect include polymorphic filtering.
+ with testing.expect_deprecated(join_aliased_dep):
+ q = (
+ s.query(lambda: User)
+ .join(lambda: User.addresses, aliased=True)
+ .filter(lambda: Address.email_address == "foo")
+ )
+ self.assert_compile(
+ q,
+ "SELECT users.id AS users_id, users.name AS users_name "
+ "FROM users JOIN addresses AS addresses_1 "
+ "ON users.id = addresses_1.user_id "
+ "WHERE addresses_1.email_address = :email_address_1",
+ )
+
def test_overlapping_paths_two(self):
User = self.classes.User
"generate_series(:generate_series_1, anon_1.bookcase_shelves) "
"AS anon_2 ON true",
)
+
+
+class ParentTest(QueryTest, AssertsCompiledSQL):
+ __dialect__ = "default"
+
+ def test_o2m(self):
+ User, orders, Order = (
+ self.classes.User,
+ self.tables.orders,
+ self.classes.Order,
+ )
+
+ sess = fixture_session()
+ q = sess.query(User)
+
+ u1 = q.filter_by(name="jack").one()
+
+ # test auto-lookup of property
+ with assertions.expect_deprecated_20(query_wparent_dep):
+ o = sess.query(Order).with_parent(u1).all()
+ assert [
+ Order(description="order 1"),
+ Order(description="order 3"),
+ Order(description="order 5"),
+ ] == o
+
+ # test with explicit property
+ with assertions.expect_deprecated_20(query_wparent_dep):
+ o = sess.query(Order).with_parent(u1, property=User.orders).all()
+ assert [
+ Order(description="order 1"),
+ Order(description="order 3"),
+ Order(description="order 5"),
+ ] == o
+
+ with assertions.expect_deprecated_20(query_wparent_dep):
+ # test generative criterion
+ o = sess.query(Order).with_parent(u1).filter(orders.c.id > 2).all()
+ assert [
+ Order(description="order 3"),
+ Order(description="order 5"),
+ ] == o
+
+ def test_select_from(self):
+ User, Address = self.classes.User, self.classes.Address
+
+ sess = fixture_session()
+ u1 = sess.query(User).get(7)
+ with assertions.expect_deprecated_20(query_wparent_dep):
+ q = sess.query(Address).select_from(Address).with_parent(u1)
+ self.assert_compile(
+ q,
+ "SELECT addresses.id AS addresses_id, "
+ "addresses.user_id AS addresses_user_id, "
+ "addresses.email_address AS addresses_email_address "
+ "FROM addresses WHERE :param_1 = addresses.user_id",
+ {"param_1": 7},
+ )
+
+ def test_from_entity_query_entity(self):
+ User, Address = self.classes.User, self.classes.Address
+
+ sess = fixture_session()
+ u1 = sess.query(User).get(7)
+ with assertions.expect_deprecated_20(query_wparent_dep):
+ q = sess.query(User, Address).with_parent(
+ u1, User.addresses, from_entity=Address
+ )
+ self.assert_compile(
+ q,
+ "SELECT users.id AS users_id, users.name AS users_name, "
+ "addresses.id AS addresses_id, addresses.user_id "
+ "AS addresses_user_id, "
+ "addresses.email_address AS addresses_email_address "
+ "FROM users, addresses "
+ "WHERE :param_1 = addresses.user_id",
+ {"param_1": 7},
+ )
+
+ def test_select_from_alias(self):
+ User, Address = self.classes.User, self.classes.Address
+
+ sess = fixture_session()
+ u1 = sess.query(User).get(7)
+ a1 = aliased(Address)
+ with assertions.expect_deprecated_20(query_wparent_dep):
+ q = sess.query(a1).with_parent(u1)
+ self.assert_compile(
+ q,
+ "SELECT addresses_1.id AS addresses_1_id, "
+ "addresses_1.user_id AS addresses_1_user_id, "
+ "addresses_1.email_address AS addresses_1_email_address "
+ "FROM addresses AS addresses_1 "
+ "WHERE :param_1 = addresses_1.user_id",
+ {"param_1": 7},
+ )
+
+ def test_select_from_alias_explicit_prop(self):
+ User, Address = self.classes.User, self.classes.Address
+
+ sess = fixture_session()
+ u1 = sess.query(User).get(7)
+ a1 = aliased(Address)
+ with assertions.expect_deprecated_20(query_wparent_dep):
+ q = sess.query(a1).with_parent(u1, User.addresses)
+ self.assert_compile(
+ q,
+ "SELECT addresses_1.id AS addresses_1_id, "
+ "addresses_1.user_id AS addresses_1_user_id, "
+ "addresses_1.email_address AS addresses_1_email_address "
+ "FROM addresses AS addresses_1 "
+ "WHERE :param_1 = addresses_1.user_id",
+ {"param_1": 7},
+ )
+
+ def test_select_from_alias_from_entity(self):
+ User, Address = self.classes.User, self.classes.Address
+
+ sess = fixture_session()
+ u1 = sess.query(User).get(7)
+ a1 = aliased(Address)
+ a2 = aliased(Address)
+ with assertions.expect_deprecated_20(query_wparent_dep):
+ q = sess.query(a1, a2).with_parent(
+ u1, User.addresses, from_entity=a2
+ )
+ self.assert_compile(
+ q,
+ "SELECT addresses_1.id AS addresses_1_id, "
+ "addresses_1.user_id AS addresses_1_user_id, "
+ "addresses_1.email_address AS addresses_1_email_address, "
+ "addresses_2.id AS addresses_2_id, "
+ "addresses_2.user_id AS addresses_2_user_id, "
+ "addresses_2.email_address AS addresses_2_email_address "
+ "FROM addresses AS addresses_1, "
+ "addresses AS addresses_2 WHERE :param_1 = addresses_2.user_id",
+ {"param_1": 7},
+ )
+
+ def test_select_from_alias_of_type(self):
+ User, Address = self.classes.User, self.classes.Address
+
+ sess = fixture_session()
+ u1 = sess.query(User).get(7)
+ a1 = aliased(Address)
+ a2 = aliased(Address)
+ with assertions.expect_deprecated_20(query_wparent_dep):
+ q = sess.query(a1, a2).with_parent(u1, User.addresses.of_type(a2))
+ self.assert_compile(
+ q,
+ "SELECT addresses_1.id AS addresses_1_id, "
+ "addresses_1.user_id AS addresses_1_user_id, "
+ "addresses_1.email_address AS addresses_1_email_address, "
+ "addresses_2.id AS addresses_2_id, "
+ "addresses_2.user_id AS addresses_2_user_id, "
+ "addresses_2.email_address AS addresses_2_email_address "
+ "FROM addresses AS addresses_1, "
+ "addresses AS addresses_2 WHERE :param_1 = addresses_2.user_id",
+ {"param_1": 7},
+ )
+
+ def test_noparent(self):
+ Item, User = self.classes.Item, self.classes.User
+
+ sess = fixture_session()
+ q = sess.query(User)
+
+ u1 = q.filter_by(name="jack").one()
+
+ with assertions.expect_deprecated_20(query_wparent_dep):
+ with assertions.expect_raises_message(
+ sa_exc.InvalidRequestError,
+ "Could not locate a property which relates "
+ "instances of class 'Item' to instances of class 'User'",
+ ):
+ q = sess.query(Item).with_parent(u1)
+
+ def test_m2m(self):
+ Item, Keyword = self.classes.Item, self.classes.Keyword
+
+ sess = fixture_session()
+ i1 = sess.query(Item).filter_by(id=2).one()
+ with assertions.expect_deprecated_20(query_wparent_dep):
+ k = sess.query(Keyword).with_parent(i1).all()
+ assert [
+ Keyword(name="red"),
+ Keyword(name="small"),
+ Keyword(name="square"),
+ ] == k
+
+ def test_with_transient(self):
+ User, Order = self.classes.User, self.classes.Order
+
+ sess = fixture_session()
+
+ q = sess.query(User)
+ u1 = q.filter_by(name="jack").one()
+ utrans = User(id=u1.id)
+ with assertions.expect_deprecated_20(query_wparent_dep):
+ o = sess.query(Order).with_parent(utrans, User.orders)
+ eq_(
+ [
+ Order(description="order 1"),
+ Order(description="order 3"),
+ Order(description="order 5"),
+ ],
+ o.all(),
+ )
+
+ def test_with_pending_autoflush(self):
+ Order, User = self.classes.Order, self.classes.User
+
+ sess = fixture_session()
+
+ o1 = sess.query(Order).first()
+ opending = Order(id=20, user_id=o1.user_id)
+ sess.add(opending)
+ with assertions.expect_deprecated_20(query_wparent_dep):
+ eq_(
+ sess.query(User).with_parent(opending, Order.user).one(),
+ User(id=o1.user_id),
+ )
+
+ def test_with_pending_no_autoflush(self):
+ Order, User = self.classes.Order, self.classes.User
+
+ sess = fixture_session(autoflush=False)
+
+ o1 = sess.query(Order).first()
+ opending = Order(user_id=o1.user_id)
+ sess.add(opending)
+ with assertions.expect_deprecated_20(query_wparent_dep):
+ eq_(
+ sess.query(User).with_parent(opending, Order.user).one(),
+ User(id=o1.user_id),
+ )
+
+ def test_unique_binds_union(self):
+ """bindparams used in the 'parent' query are unique"""
+ User, Address = self.classes.User, self.classes.Address
+
+ sess = fixture_session()
+ u1, u2 = sess.query(User).order_by(User.id)[0:2]
+
+ with assertions.expect_deprecated_20(query_wparent_dep):
+ q1 = sess.query(Address).with_parent(u1, User.addresses)
+ with assertions.expect_deprecated_20(query_wparent_dep):
+ q2 = sess.query(Address).with_parent(u2, User.addresses)
+
+ self.assert_compile(
+ q1.union(q2),
+ "SELECT anon_1.addresses_id AS anon_1_addresses_id, "
+ "anon_1.addresses_user_id AS anon_1_addresses_user_id, "
+ "anon_1.addresses_email_address AS "
+ "anon_1_addresses_email_address FROM (SELECT addresses.id AS "
+ "addresses_id, addresses.user_id AS addresses_user_id, "
+ "addresses.email_address AS addresses_email_address FROM "
+ "addresses WHERE :param_1 = addresses.user_id UNION SELECT "
+ "addresses.id AS addresses_id, addresses.user_id AS "
+ "addresses_user_id, addresses.email_address "
+ "AS addresses_email_address "
+ "FROM addresses WHERE :param_2 = addresses.user_id) AS anon_1",
+ checkparams={"param_1": 7, "param_2": 8},
+ )
):
self.assert_sql_count(testing.db, fn, 2)
- def test_does_filter_aliasing_work(self, plain_fixture):
- User, Address = plain_fixture
-
- s = Session(testing.db, future=True)
-
- # aliased=True is to be deprecated, other filter lambdas
- # that go into effect include polymorphic filtering.
- q = (
- s.query(lambda: User)
- .join(lambda: User.addresses, aliased=True)
- .filter(lambda: Address.email_address == "foo")
- )
- self.assert_compile(
- q,
- "SELECT users.id AS users_id, users.name AS users_name "
- "FROM users JOIN addresses AS addresses_1 "
- "ON users.id = addresses_1.user_id "
- "WHERE addresses_1.email_address = :email_address_1",
- )
-
@testing.combinations(
lambda s, User, Address: s.query(lambda: User).join(lambda: Address),
lambda s, User, Address: s.query(lambda: User).join(
from sqlalchemy.orm import exc as orm_exc
from sqlalchemy.orm import relationship
from sqlalchemy.orm import Session
+from sqlalchemy.orm import with_parent
from sqlalchemy.testing import assert_raises
from sqlalchemy.testing import eq_
from sqlalchemy.testing import fixtures
[Order(id=1), Order(id=5)],
fixture_session()
.query(closed_mapper)
- .with_parent(user, property=User.closed_orders)
+ .filter(with_parent(user, User.closed_orders))
.all(),
)
eq_(
[Order(id=3)],
fixture_session()
.query(open_mapper)
- .with_parent(user, property=User.open_orders)
+ .filter(with_parent(user, User.open_orders))
.all(),
)
from sqlalchemy import String
from sqlalchemy import testing
from sqlalchemy import TypeDecorator
+from sqlalchemy.orm import make_transient
from sqlalchemy.orm import relationship
-from sqlalchemy.orm.session import make_transient
+from sqlalchemy.orm import with_parent
from sqlalchemy.testing import assert_raises
from sqlalchemy.testing import assert_raises_message
from sqlalchemy.testing import eq_
u2 = sess.query(User).get(u2.username)
u2.username = "wendy"
sess.flush()
- r = sess.query(Item).with_parent(u2).all()
+ r = sess.query(Item).filter(with_parent(u2, User.items)).all()
eq_(Item(itemname="item2"), r[0])
def test_manytoone_deferred_relationship_expr(self):
__dialect__ = "default"
def test_o2m(self):
- User, orders, Order = (
+ User, Order = (
self.classes.User,
- self.tables.orders,
self.classes.Order,
)
u1 = q.filter_by(name="jack").one()
- # test auto-lookup of property
- o = sess.query(Order).with_parent(u1).all()
- assert [
- Order(description="order 1"),
- Order(description="order 3"),
- Order(description="order 5"),
- ] == o
-
- # test with explicit property
- o = sess.query(Order).with_parent(u1, property=User.orders).all()
- assert [
- Order(description="order 1"),
- Order(description="order 3"),
- Order(description="order 5"),
- ] == o
-
o = sess.query(Order).filter(with_parent(u1, User.orders)).all()
assert [
Order(description="order 1"),
Order(description="order 5"),
] == o
- # test generative criterion
- o = sess.query(Order).with_parent(u1).filter(orders.c.id > 2).all()
- assert [
- Order(description="order 3"),
- Order(description="order 5"),
- ] == o
-
- # test against None for parent? this can't be done with the current
- # API since we don't know what mapper to use
- # assert
- # sess.query(Order).with_parent(None, property='addresses').all()
- # == [Order(description="order 5")]
-
def test_select_from(self):
User, Address = self.classes.User, self.classes.Address
sess = fixture_session()
u1 = sess.query(User).get(7)
- q = sess.query(Address).select_from(Address).with_parent(u1)
+ q = (
+ sess.query(Address)
+ .select_from(Address)
+ .filter(with_parent(u1, User.addresses))
+ )
self.assert_compile(
q,
"SELECT addresses.id AS addresses_id, "
{"param_1": 7},
)
- def test_from_entity_query_entity(self):
- User, Address = self.classes.User, self.classes.Address
-
- sess = fixture_session()
- u1 = sess.query(User).get(7)
- q = sess.query(User, Address).with_parent(
- u1, User.addresses, from_entity=Address
- )
- self.assert_compile(
- q,
- "SELECT users.id AS users_id, users.name AS users_name, "
- "addresses.id AS addresses_id, addresses.user_id "
- "AS addresses_user_id, "
- "addresses.email_address AS addresses_email_address "
- "FROM users, addresses "
- "WHERE :param_1 = addresses.user_id",
- {"param_1": 7},
- )
-
def test_select_from_alias(self):
User, Address = self.classes.User, self.classes.Address
sess = fixture_session()
u1 = sess.query(User).get(7)
a1 = aliased(Address)
- q = sess.query(a1).with_parent(u1)
- self.assert_compile(
- q,
- "SELECT addresses_1.id AS addresses_1_id, "
- "addresses_1.user_id AS addresses_1_user_id, "
- "addresses_1.email_address AS addresses_1_email_address "
- "FROM addresses AS addresses_1 "
- "WHERE :param_1 = addresses_1.user_id",
- {"param_1": 7},
- )
-
- def test_select_from_alias_explicit_prop(self):
- User, Address = self.classes.User, self.classes.Address
-
- sess = fixture_session()
- u1 = sess.query(User).get(7)
- a1 = aliased(Address)
- q = sess.query(a1).with_parent(u1, User.addresses)
+ q = sess.query(a1).filter(with_parent(u1, User.addresses.of_type(a1)))
self.assert_compile(
q,
"SELECT addresses_1.id AS addresses_1_id, "
u1 = sess.query(User).get(7)
a1 = aliased(Address)
a2 = aliased(Address)
- q = sess.query(a1, a2).with_parent(u1, User.addresses, from_entity=a2)
+ q = sess.query(a1, a2).filter(
+ with_parent(u1, User.addresses, from_entity=a2)
+ )
self.assert_compile(
q,
"SELECT addresses_1.id AS addresses_1_id, "
u1 = sess.query(User).get(7)
a1 = aliased(Address)
a2 = aliased(Address)
- q = sess.query(a1, a2).with_parent(u1, User.addresses.of_type(a2))
+ q = sess.query(a1, a2).filter(
+ with_parent(u1, User.addresses.of_type(a2))
+ )
self.assert_compile(
q,
"SELECT addresses_1.id AS addresses_1_id, "
u1 = q.filter_by(name="jack").one()
- try:
- q = sess.query(Item).with_parent(u1)
- assert False
- except sa_exc.InvalidRequestError as e:
- assert (
- str(e) == "Could not locate a property which relates "
- "instances of class 'Item' to instances of class 'User'"
- )
+ # TODO: this can perhaps raise an error, then again it's doing what's
+ # asked...
+ q = sess.query(Item).filter(with_parent(u1, User.orders))
+ self.assert_compile(
+ q,
+ "SELECT items.id AS items_id, "
+ "items.description AS items_description "
+ "FROM items, orders WHERE :param_1 = orders.user_id",
+ )
def test_m2m(self):
Item, Keyword = self.classes.Item, self.classes.Keyword
sess = fixture_session()
i1 = sess.query(Item).filter_by(id=2).one()
- k = sess.query(Keyword).with_parent(i1).all()
- assert [
- Keyword(name="red"),
- Keyword(name="small"),
- Keyword(name="square"),
- ] == k
+ k = sess.query(Keyword).filter(with_parent(i1, Item.keywords)).all()
+ eq_(
+ k,
+ [
+ Keyword(name="red"),
+ Keyword(name="small"),
+ Keyword(name="square"),
+ ],
+ )
def test_with_transient(self):
User, Order = self.classes.User, self.classes.Order
q = sess.query(User)
u1 = q.filter_by(name="jack").one()
utrans = User(id=u1.id)
- o = sess.query(Order).with_parent(utrans, User.orders)
- eq_(
- [
- Order(description="order 1"),
- Order(description="order 3"),
- Order(description="order 5"),
- ],
- o.all(),
- )
-
o = sess.query(Order).filter(with_parent(utrans, User.orders))
eq_(
[
o1 = sess.query(Order).first()
opending = Order(id=20, user_id=o1.user_id)
sess.add(opending)
- eq_(
- sess.query(User).with_parent(opending, Order.user).one(),
- User(id=o1.user_id),
- )
eq_(
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, Order.user).one(),
+ sess.query(User).filter(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, User.addresses)
- q2 = sess.query(Address).with_parent(u2, User.addresses)
+ q1 = sess.query(Address).filter(with_parent(u1, User.addresses))
+ q2 = sess.query(Address).filter(with_parent(u2, User.addresses))
self.assert_compile(
q1.union(q2),
sess = fixture_session()
- q = sess.query(User).with_parent(Address(user_id=None), Address.user)
+ q = sess.query(User).filter(
+ with_parent(Address(user_id=None), Address.user)
+ )
with expect_warnings("Got None for value of column"):
self.assert_compile(
q,
User, Address = self.classes.User, self.classes.Address
s = fixture_session()
- q = s.query(User).with_parent(
- Address(user_id=None, email_address=None), Address.special_user
+ q = s.query(User).filter(
+ with_parent(
+ Address(user_id=None, email_address=None), Address.special_user
+ )
)
with expect_warnings("Got None for value of column"):
u1 = q.filter_by(**{nameprop: "jack"}).one()
- o = sess.query(Order).with_parent(u1, property=orderprop).all()
+ o = sess.query(Order).filter(with_parent(u1, orderprop)).all()
assert [
Order(description="order 1"),
Order(description="order 3"),
self._descendants_fixture(data=False)
Entity = self.classes.Entity
sess = fixture_session()
+
+ da = aliased(Entity)
self.assert_compile(
- sess.query(Entity).join(Entity.descendants, aliased=True),
+ sess.query(Entity).join(Entity.descendants.of_type(da)),
"SELECT entity.path AS entity_path FROM entity JOIN entity AS "
"entity_1 ON entity_1.path LIKE entity.path || :path_1",
)
def _test_join_aliasing(self, sess):
Employee = self.classes.Employee
+ ea = aliased(Employee)
eq_(
[
n
for n, in sess.query(Employee.name)
- .join(Employee.reports_to, aliased=True)
- .filter_by(name="emp5")
- .reset_joinpoint()
+ .join(Employee.reports_to.of_type(ea))
+ .filter(ea.name == "emp5")
+ # broken until #7244 is fixed due to of_type() usage
+ # .filter_by(name="emp5")
.order_by(Employee.name)
],
["emp6", "emp7"],
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
