import sqlalchemy as sa from sqlalchemy import ForeignKey from sqlalchemy import Integer from sqlalchemy import testing from sqlalchemy import util from sqlalchemy.orm import aliased from sqlalchemy.orm import contains_eager from sqlalchemy.orm import create_session from sqlalchemy.orm import defaultload from sqlalchemy.orm import defer from sqlalchemy.orm import deferred 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 query_expression from sqlalchemy.orm import relationship from sqlalchemy.orm import Session from sqlalchemy.orm import subqueryload from sqlalchemy.orm import synonym from sqlalchemy.orm import undefer from sqlalchemy.orm import undefer_group from sqlalchemy.orm import with_expression from sqlalchemy.orm import with_polymorphic from sqlalchemy.testing import assert_raises_message from sqlalchemy.testing import AssertsCompiledSQL from sqlalchemy.testing import eq_ from sqlalchemy.testing import fixtures from sqlalchemy.testing.schema import Column from test.orm import _fixtures from .inheritance._poly_fixtures import _Polymorphic from .inheritance._poly_fixtures import Boss from .inheritance._poly_fixtures import Company from .inheritance._poly_fixtures import Manager from .inheritance._poly_fixtures import Person class DeferredTest(AssertsCompiledSQL, _fixtures.FixtureTest): def test_basic(self): """A basic deferred load.""" Order, orders = self.classes.Order, self.tables.orders mapper( Order, orders, properties={"description": deferred(orders.c.description)}, ) o = Order() self.assert_(o.description is None) q = create_session().query(Order).order_by(Order.id) def go(): result = q.all() o2 = result[2] x = o2.description self.sql_eq_( go, [ ( "SELECT orders.id AS orders_id, " "orders.user_id AS orders_user_id, " "orders.address_id AS orders_address_id, " "orders.isopen AS orders_isopen " "FROM orders ORDER BY orders.id", {}, ), ( "SELECT orders.description AS orders_description " "FROM orders WHERE orders.id = :param_1", {"param_1": 3}, ), ], ) def test_defer_primary_key(self): """what happens when we try to defer the primary key?""" Order, orders = self.classes.Order, self.tables.orders mapper(Order, orders, properties={"id": deferred(orders.c.id)}) # right now, it's not that graceful :) q = create_session().query(Order) assert_raises_message( sa.exc.NoSuchColumnError, "Could not locate", q.first ) def test_unsaved(self): """Deferred loading does not kick in when just PK cols are set.""" Order, orders = self.classes.Order, self.tables.orders mapper( Order, orders, properties={"description": deferred(orders.c.description)}, ) sess = create_session() o = Order() sess.add(o) o.id = 7 def go(): o.description = "some description" self.sql_count_(0, go) def test_synonym_group_bug(self): orders, Order = self.tables.orders, self.classes.Order mapper( Order, orders, properties={ "isopen": synonym("_isopen", map_column=True), "description": deferred(orders.c.description, group="foo"), }, ) sess = create_session() o1 = sess.query(Order).get(1) eq_(o1.description, "order 1") def test_unsaved_2(self): Order, orders = self.classes.Order, self.tables.orders mapper( Order, orders, properties={"description": deferred(orders.c.description)}, ) sess = create_session() o = Order() sess.add(o) def go(): o.description = "some description" self.sql_count_(0, go) def test_unsaved_group(self): """Deferred loading doesn't kick in when just PK cols are set""" orders, Order = self.tables.orders, self.classes.Order mapper( Order, orders, properties=dict( description=deferred(orders.c.description, group="primary"), opened=deferred(orders.c.isopen, group="primary"), ), ) sess = create_session() o = Order() sess.add(o) o.id = 7 def go(): o.description = "some description" self.sql_count_(0, go) def test_unsaved_group_2(self): orders, Order = self.tables.orders, self.classes.Order mapper( Order, orders, properties=dict( description=deferred(orders.c.description, group="primary"), opened=deferred(orders.c.isopen, group="primary"), ), ) sess = create_session() o = Order() sess.add(o) def go(): o.description = "some description" self.sql_count_(0, go) def test_save(self): Order, orders = self.classes.Order, self.tables.orders m = mapper( Order, orders, properties={"description": deferred(orders.c.description)}, ) sess = create_session() o2 = sess.query(Order).get(2) o2.isopen = 1 sess.flush() def test_group(self): """Deferred load with a group""" orders, Order = self.tables.orders, self.classes.Order mapper( Order, orders, properties=util.OrderedDict( [ ("userident", deferred(orders.c.user_id, group="primary")), ( "addrident", deferred(orders.c.address_id, group="primary"), ), ( "description", deferred(orders.c.description, group="primary"), ), ("opened", deferred(orders.c.isopen, group="primary")), ] ), ) sess = create_session() q = sess.query(Order).order_by(Order.id) def go(): result = q.all() o2 = result[2] eq_(o2.opened, 1) eq_(o2.userident, 7) eq_(o2.description, "order 3") self.sql_eq_( go, [ ( "SELECT orders.id AS orders_id " "FROM orders ORDER BY orders.id", {}, ), ( "SELECT orders.user_id AS orders_user_id, " "orders.address_id AS orders_address_id, " "orders.description AS orders_description, " "orders.isopen AS orders_isopen " "FROM orders WHERE orders.id = :param_1", {"param_1": 3}, ), ], ) o2 = q.all()[2] eq_(o2.description, "order 3") assert o2 not in sess.dirty o2.description = "order 3" def go(): sess.flush() self.sql_count_(0, go) def test_preserve_changes(self): """A deferred load operation doesn't revert modifications on attributes """ orders, Order = self.tables.orders, self.classes.Order mapper( Order, orders, properties={ "userident": deferred(orders.c.user_id, group="primary"), "description": deferred(orders.c.description, group="primary"), "opened": deferred(orders.c.isopen, group="primary"), }, ) sess = create_session() o = sess.query(Order).get(3) assert "userident" not in o.__dict__ o.description = "somenewdescription" eq_(o.description, "somenewdescription") def go(): eq_(o.opened, 1) self.assert_sql_count(testing.db, go, 1) eq_(o.description, "somenewdescription") assert o in sess.dirty def test_commits_state(self): """ When deferred elements are loaded via a group, they get the proper CommittedState and don't result in changes being committed """ orders, Order = self.tables.orders, self.classes.Order mapper( Order, orders, properties={ "userident": deferred(orders.c.user_id, group="primary"), "description": deferred(orders.c.description, group="primary"), "opened": deferred(orders.c.isopen, group="primary"), }, ) sess = create_session() o2 = sess.query(Order).get(3) # this will load the group of attributes eq_(o2.description, "order 3") assert o2 not in sess.dirty # this will mark it as 'dirty', but nothing actually changed o2.description = "order 3" # therefore the flush() shouldn't actually issue any SQL self.assert_sql_count(testing.db, sess.flush, 0) def test_map_selectable_wo_deferred(self): """test mapping to a selectable with deferred cols, the selectable doesn't include the deferred col. """ Order, orders = self.classes.Order, self.tables.orders order_select = sa.select( [ orders.c.id, orders.c.user_id, orders.c.address_id, orders.c.description, orders.c.isopen, ] ).alias() mapper( Order, order_select, properties={"description": deferred(order_select.c.description)}, ) sess = Session() o1 = sess.query(Order).order_by(Order.id).first() assert "description" not in o1.__dict__ eq_(o1.description, "order 1") class DeferredOptionsTest(AssertsCompiledSQL, _fixtures.FixtureTest): __dialect__ = "default" def test_options(self): """Options on a mapper to create deferred and undeferred columns""" orders, Order = self.tables.orders, self.classes.Order mapper(Order, orders) sess = create_session() q = sess.query(Order).order_by(Order.id).options(defer("user_id")) def go(): q.all()[0].user_id self.sql_eq_( go, [ ( "SELECT orders.id AS orders_id, " "orders.address_id AS orders_address_id, " "orders.description AS orders_description, " "orders.isopen AS orders_isopen " "FROM orders ORDER BY orders.id", {}, ), ( "SELECT orders.user_id AS orders_user_id " "FROM orders WHERE orders.id = :param_1", {"param_1": 1}, ), ], ) sess.expunge_all() q2 = q.options(undefer("user_id")) self.sql_eq_( q2.all, [ ( "SELECT orders.id AS orders_id, " "orders.user_id AS orders_user_id, " "orders.address_id AS orders_address_id, " "orders.description AS orders_description, " "orders.isopen AS orders_isopen " "FROM orders ORDER BY orders.id", {}, ) ], ) def test_undefer_group(self): orders, Order = self.tables.orders, self.classes.Order mapper( Order, orders, properties=util.OrderedDict( [ ("userident", deferred(orders.c.user_id, group="primary")), ( "description", deferred(orders.c.description, group="primary"), ), ("opened", deferred(orders.c.isopen, group="primary")), ] ), ) sess = create_session() q = sess.query(Order).order_by(Order.id) def go(): result = q.options(undefer_group("primary")).all() o2 = result[2] eq_(o2.opened, 1) eq_(o2.userident, 7) eq_(o2.description, "order 3") self.sql_eq_( go, [ ( "SELECT orders.user_id AS orders_user_id, " "orders.description AS orders_description, " "orders.isopen AS orders_isopen, " "orders.id AS orders_id, " "orders.address_id AS orders_address_id " "FROM orders ORDER BY orders.id", {}, ) ], ) def test_undefer_group_multi(self): orders, Order = self.tables.orders, self.classes.Order mapper( Order, orders, properties=util.OrderedDict( [ ("userident", deferred(orders.c.user_id, group="primary")), ( "description", deferred(orders.c.description, group="primary"), ), ("opened", deferred(orders.c.isopen, group="secondary")), ] ), ) sess = create_session() q = sess.query(Order).order_by(Order.id) def go(): result = q.options( undefer_group("primary"), undefer_group("secondary") ).all() o2 = result[2] eq_(o2.opened, 1) eq_(o2.userident, 7) eq_(o2.description, "order 3") self.sql_eq_( go, [ ( "SELECT orders.user_id AS orders_user_id, " "orders.description AS orders_description, " "orders.isopen AS orders_isopen, " "orders.id AS orders_id, " "orders.address_id AS orders_address_id " "FROM orders ORDER BY orders.id", {}, ) ], ) def test_undefer_group_multi_pathed(self): orders, Order = self.tables.orders, self.classes.Order mapper( Order, orders, properties=util.OrderedDict( [ ("userident", deferred(orders.c.user_id, group="primary")), ( "description", deferred(orders.c.description, group="primary"), ), ("opened", deferred(orders.c.isopen, group="secondary")), ] ), ) sess = create_session() q = sess.query(Order).order_by(Order.id) def go(): result = q.options( Load(Order).undefer_group("primary").undefer_group("secondary") ).all() o2 = result[2] eq_(o2.opened, 1) eq_(o2.userident, 7) eq_(o2.description, "order 3") self.sql_eq_( go, [ ( "SELECT orders.user_id AS orders_user_id, " "orders.description AS orders_description, " "orders.isopen AS orders_isopen, " "orders.id AS orders_id, " "orders.address_id AS orders_address_id " "FROM orders ORDER BY orders.id", {}, ) ], ) def test_undefer_group_from_relationship_lazyload(self): users, Order, User, orders = ( self.tables.users, self.classes.Order, self.classes.User, self.tables.orders, ) mapper( User, users, properties=dict(orders=relationship(Order, order_by=orders.c.id)), ) mapper( Order, orders, properties=util.OrderedDict( [ ("userident", deferred(orders.c.user_id, group="primary")), ( "description", deferred(orders.c.description, group="primary"), ), ("opened", deferred(orders.c.isopen, group="primary")), ] ), ) sess = create_session() q = ( sess.query(User) .filter(User.id == 7) .options(defaultload(User.orders).undefer_group("primary")) ) def go(): result = q.all() o2 = result[0].orders[1] eq_(o2.opened, 1) eq_(o2.userident, 7) eq_(o2.description, "order 3") self.sql_eq_( go, [ ( "SELECT users.id AS users_id, users.name AS users_name " "FROM users WHERE users.id = :id_1", {"id_1": 7}, ), ( "SELECT orders.user_id AS orders_user_id, " "orders.description " "AS orders_description, orders.isopen AS orders_isopen, " "orders.id AS orders_id, orders.address_id " "AS orders_address_id " "FROM orders WHERE :param_1 = orders.user_id " "ORDER BY orders.id", {"param_1": 7}, ), ], ) def test_undefer_group_from_relationship_subqueryload(self): users, Order, User, orders = ( self.tables.users, self.classes.Order, self.classes.User, self.tables.orders, ) mapper( User, users, properties=dict(orders=relationship(Order, order_by=orders.c.id)), ) mapper( Order, orders, properties=util.OrderedDict( [ ("userident", deferred(orders.c.user_id, group="primary")), ( "description", deferred(orders.c.description, group="primary"), ), ("opened", deferred(orders.c.isopen, group="primary")), ] ), ) sess = create_session() q = ( sess.query(User) .filter(User.id == 7) .options(subqueryload(User.orders).undefer_group("primary")) ) def go(): result = q.all() o2 = result[0].orders[1] eq_(o2.opened, 1) eq_(o2.userident, 7) eq_(o2.description, "order 3") self.sql_eq_( go, [ ( "SELECT users.id AS users_id, users.name AS users_name " "FROM users WHERE users.id = :id_1", {"id_1": 7}, ), ( "SELECT orders.user_id AS orders_user_id, " "orders.description " "AS orders_description, orders.isopen AS orders_isopen, " "orders.id AS orders_id, " "orders.address_id AS orders_address_id, " "anon_1.users_id AS anon_1_users_id " "FROM (SELECT users.id AS " "users_id FROM users WHERE users.id = :id_1) AS anon_1 " "JOIN orders ON anon_1.users_id = orders.user_id ORDER BY " "anon_1.users_id, orders.id", [{"id_1": 7}], ), ], ) def test_undefer_group_from_relationship_joinedload(self): users, Order, User, orders = ( self.tables.users, self.classes.Order, self.classes.User, self.tables.orders, ) mapper( User, users, properties=dict(orders=relationship(Order, order_by=orders.c.id)), ) mapper( Order, orders, properties=util.OrderedDict( [ ("userident", deferred(orders.c.user_id, group="primary")), ( "description", deferred(orders.c.description, group="primary"), ), ("opened", deferred(orders.c.isopen, group="primary")), ] ), ) sess = create_session() q = ( sess.query(User) .filter(User.id == 7) .options(joinedload(User.orders).undefer_group("primary")) ) def go(): result = q.all() o2 = result[0].orders[1] eq_(o2.opened, 1) eq_(o2.userident, 7) eq_(o2.description, "order 3") self.sql_eq_( go, [ ( "SELECT users.id AS users_id, users.name AS users_name, " "orders_1.user_id AS orders_1_user_id, " "orders_1.description AS orders_1_description, " "orders_1.isopen AS orders_1_isopen, " "orders_1.id AS orders_1_id, orders_1.address_id AS " "orders_1_address_id FROM users " "LEFT OUTER JOIN orders AS orders_1 ON users.id = " "orders_1.user_id WHERE users.id = :id_1 " "ORDER BY orders_1.id", {"id_1": 7}, ) ], ) def test_undefer_group_from_relationship_joinedload_colexpr(self): users, Order, User, orders = ( self.tables.users, self.classes.Order, self.classes.User, self.tables.orders, ) mapper( User, users, properties=dict(orders=relationship(Order, order_by=orders.c.id)), ) mapper( Order, orders, properties=util.OrderedDict( [ ("userident", deferred(orders.c.user_id, group="primary")), ( "lower_desc", deferred( sa.func.lower(orders.c.description).label(None), group="primary", ), ), ("opened", deferred(orders.c.isopen, group="primary")), ] ), ) sess = create_session() q = ( sess.query(User) .filter(User.id == 7) .options(joinedload(User.orders).undefer_group("primary")) ) def go(): result = q.all() o2 = result[0].orders[1] eq_(o2.opened, 1) eq_(o2.userident, 7) eq_(o2.lower_desc, "order 3") self.sql_eq_( go, [ ( "SELECT users.id AS users_id, users.name AS users_name, " "orders_1.user_id AS orders_1_user_id, " "lower(orders_1.description) AS lower_1, " "orders_1.isopen AS orders_1_isopen, " "orders_1.id AS orders_1_id, " "orders_1.address_id AS orders_1_address_id, " "orders_1.description AS orders_1_description FROM users " "LEFT OUTER JOIN orders AS orders_1 ON users.id = " "orders_1.user_id WHERE users.id = :id_1 " "ORDER BY orders_1.id", {"id_1": 7}, ) ], ) def test_undefer_star(self): orders, Order = self.tables.orders, self.classes.Order mapper( Order, orders, properties=util.OrderedDict( [ ("userident", deferred(orders.c.user_id)), ("description", deferred(orders.c.description)), ("opened", deferred(orders.c.isopen)), ] ), ) sess = create_session() q = sess.query(Order).options(Load(Order).undefer("*")) self.assert_compile( q, "SELECT orders.user_id AS orders_user_id, " "orders.description AS orders_description, " "orders.isopen AS orders_isopen, " "orders.id AS orders_id, " "orders.address_id AS orders_address_id " "FROM orders", ) def test_locates_col(self): """changed in 1.0 - we don't search for deferred cols in the result now. """ orders, Order = self.tables.orders, self.classes.Order mapper( Order, orders, properties={"description": deferred(orders.c.description)}, ) sess = create_session() o1 = ( sess.query(Order) .order_by(Order.id) .add_column(orders.c.description) .first() )[0] def go(): eq_(o1.description, "order 1") # prior to 1.0 we'd search in the result for this column # self.sql_count_(0, go) self.sql_count_(1, go) def test_locates_col_rowproc_only(self): """changed in 1.0 - we don't search for deferred cols in the result now. Because the loading for ORM Query and Query from a core select is now split off, we test loading from a plain select() separately. """ orders, Order = self.tables.orders, self.classes.Order mapper( Order, orders, properties={"description": deferred(orders.c.description)}, ) sess = create_session() stmt = sa.select([Order]).order_by(Order.id) o1 = (sess.query(Order).from_statement(stmt).all())[0] def go(): eq_(o1.description, "order 1") # prior to 1.0 we'd search in the result for this column # self.sql_count_(0, go) self.sql_count_(1, go) 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, ) mapper( Item, items, properties=dict(description=deferred(items.c.description)), ) mapper( Order, orders, properties=dict(items=relationship(Item, secondary=order_items)), ) mapper( User, users, properties=dict(orders=relationship(Order, order_by=orders.c.id)), ) sess = create_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() result = q.options(undefer("orders.items.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") def test_path_entity(self): """test the legacy *addl_attrs argument.""" User = self.classes.User Order = self.classes.Order Item = self.classes.Item users = self.tables.users orders = self.tables.orders items = self.tables.items order_items = self.tables.order_items mapper( User, users, properties={"orders": relationship(Order, lazy="joined")}, ) mapper( Order, orders, properties={ "items": relationship( Item, secondary=order_items, lazy="joined" ) }, ) mapper(Item, items) sess = create_session() exp = ( "SELECT users.id AS users_id, users.name AS users_name, " "items_1.id AS items_1_id, orders_1.id AS orders_1_id, " "orders_1.user_id AS orders_1_user_id, orders_1.address_id " "AS orders_1_address_id, orders_1.description AS " "orders_1_description, orders_1.isopen AS orders_1_isopen " "FROM users LEFT OUTER JOIN orders AS orders_1 " "ON users.id = orders_1.user_id LEFT OUTER JOIN " "(order_items AS order_items_1 JOIN items AS items_1 " "ON items_1.id = order_items_1.item_id) " "ON orders_1.id = order_items_1.order_id" ) q = sess.query(User).options( defer(User.orders, Order.items, Item.description) ) self.assert_compile(q, exp) def test_chained_multi_col_options(self): users, User = self.tables.users, self.classes.User orders, Order = self.tables.orders, self.classes.Order mapper(User, users, properties={"orders": relationship(Order)}) mapper(Order, orders) sess = create_session() q = sess.query(User).options( joinedload(User.orders).defer("description").defer("isopen") ) self.assert_compile( q, "SELECT users.id AS users_id, " "users.name AS users_name, " "orders_1.id AS orders_1_id, " "orders_1.user_id AS orders_1_user_id, " "orders_1.address_id AS orders_1_address_id " "FROM users " "LEFT OUTER JOIN orders AS orders_1 " "ON users.id = orders_1.user_id", ) def test_load_only_no_pk(self): orders, Order = self.tables.orders, self.classes.Order mapper(Order, orders) sess = create_session() q = sess.query(Order).options(load_only("isopen", "description")) self.assert_compile( q, "SELECT orders.id AS orders_id, " "orders.description AS orders_description, " "orders.isopen AS orders_isopen FROM orders", ) def test_load_only_no_pk_rt(self): orders, Order = self.tables.orders, self.classes.Order mapper(Order, orders) sess = create_session() q = ( sess.query(Order) .order_by(Order.id) .options(load_only("isopen", "description")) ) eq_(q.first(), Order(id=1)) def test_load_only_w_deferred(self): orders, Order = self.tables.orders, self.classes.Order mapper( Order, orders, properties={"description": deferred(orders.c.description)}, ) sess = create_session() q = sess.query(Order).options( load_only("isopen", "description"), undefer("user_id") ) self.assert_compile( q, "SELECT orders.description AS orders_description, " "orders.id AS orders_id, " "orders.user_id AS orders_user_id, " "orders.isopen AS orders_isopen FROM orders", ) def test_load_only_propagate_unbound(self): self._test_load_only_propagate(False) def test_load_only_propagate_bound(self): self._test_load_only_propagate(True) def _test_load_only_propagate(self, use_load): User = self.classes.User Address = self.classes.Address users = self.tables.users addresses = self.tables.addresses mapper(User, users, properties={"addresses": relationship(Address)}) mapper(Address, addresses) sess = create_session() expected = [ ( "SELECT users.id AS users_id, users.name AS users_name " "FROM users WHERE users.id IN (:id_1, :id_2)", {"id_2": 8, "id_1": 7}, ), ( "SELECT addresses.id AS addresses_id, " "addresses.email_address AS addresses_email_address " "FROM addresses WHERE :param_1 = addresses.user_id", {"param_1": 7}, ), ( "SELECT addresses.id AS addresses_id, " "addresses.email_address AS addresses_email_address " "FROM addresses WHERE :param_1 = addresses.user_id", {"param_1": 8}, ), ] if use_load: opt = ( Load(User) .defaultload(User.addresses) .load_only("id", "email_address") ) else: opt = defaultload(User.addresses).load_only("id", "email_address") q = sess.query(User).options(opt).filter(User.id.in_([7, 8])) def go(): for user in q: user.addresses self.sql_eq_(go, expected) def test_load_only_parent_specific(self): User = self.classes.User Address = self.classes.Address Order = self.classes.Order users = self.tables.users addresses = self.tables.addresses orders = self.tables.orders mapper(User, users) mapper(Address, addresses) mapper(Order, orders) sess = create_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"), ) self.assert_compile( q, "SELECT users.id AS users_id, " "users.name AS users_name, " "orders.id AS orders_id, " "addresses.id AS addresses_id, " "addresses.email_address AS addresses_email_address " "FROM users, orders, addresses", ) def test_load_only_path_specific(self): User = self.classes.User Address = self.classes.Address Order = self.classes.Order users = self.tables.users addresses = self.tables.addresses orders = self.tables.orders mapper( User, users, properties=util.OrderedDict( [ ("addresses", relationship(Address, lazy="joined")), ("orders", relationship(Order, lazy="joined")), ] ), ) mapper(Address, addresses) mapper(Order, orders) sess = create_session() q = sess.query(User).options( load_only("name") .defaultload("addresses") .load_only("id", "email_address"), defaultload("orders").load_only("id"), ) # hmmmm joinedload seems to be forcing users.id into here... self.assert_compile( q, "SELECT users.id AS users_id, users.name AS users_name, " "addresses_1.id AS addresses_1_id, " "addresses_1.email_address AS addresses_1_email_address, " "orders_1.id AS orders_1_id FROM users " "LEFT OUTER JOIN addresses AS addresses_1 " "ON users.id = addresses_1.user_id " "LEFT OUTER JOIN orders AS orders_1 " "ON users.id = orders_1.user_id", ) class SelfReferentialMultiPathTest(testing.fixtures.DeclarativeMappedTest): """test for [ticket:3822]""" @classmethod def setup_classes(cls): Base = cls.DeclarativeBasic class Node(Base): __tablename__ = "node" id = sa.Column(sa.Integer, primary_key=True) parent_id = sa.Column(sa.ForeignKey("node.id")) parent = relationship("Node", remote_side=[id]) name = sa.Column(sa.String(10)) @classmethod def insert_data(cls): Node = cls.classes.Node session = Session() session.add_all( [ Node(id=1, name="name"), Node(id=2, parent_id=1, name="name"), Node(id=3, parent_id=1, name="name"), ] ) session.commit() def test_present_overrides_deferred(self): Node = self.classes.Node session = Session() q = session.query(Node).options( joinedload(Node.parent).load_only(Node.id, Node.parent_id) ) # Node #1 will appear first as Node.parent and have # deferred applied to Node.name. it will then appear # as Node in the last row and "name" should be populated. nodes = q.order_by(Node.id.desc()).all() def go(): for node in nodes: eq_(node.name, "name") self.assert_sql_count(testing.db, go, 0) class InheritanceTest(_Polymorphic): __dialect__ = "default" def test_load_only_subclass(self): s = Session() q = ( s.query(Manager) .order_by(Manager.person_id) .options(load_only("status", "manager_name")) ) self.assert_compile( q, "SELECT managers.person_id AS managers_person_id, " "people.person_id AS people_person_id, " "people.type AS people_type, " "managers.status AS managers_status, " "managers.manager_name AS managers_manager_name " "FROM people JOIN managers " "ON people.person_id = managers.person_id " "ORDER BY managers.person_id", ) def test_load_only_subclass_and_superclass(self): s = Session() q = ( s.query(Boss) .order_by(Person.person_id) .options(load_only("status", "manager_name")) ) self.assert_compile( q, "SELECT managers.person_id AS managers_person_id, " "people.person_id AS people_person_id, " "people.type AS people_type, " "managers.status AS managers_status, " "managers.manager_name AS managers_manager_name " "FROM people JOIN managers " "ON people.person_id = managers.person_id JOIN boss " "ON managers.person_id = boss.boss_id ORDER BY people.person_id", ) def test_load_only_alias_subclass(self): s = Session() m1 = aliased(Manager, flat=True) q = ( s.query(m1) .order_by(m1.person_id) .options(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_subclass_from_relationship_polymorphic(self): s = Session() wp = with_polymorphic(Person, [Manager], flat=True) q = ( s.query(Company) .join(Company.employees.of_type(wp)) .options( contains_eager(Company.employees.of_type(wp)).load_only( wp.Manager.status, wp.Manager.manager_name ) ) ) self.assert_compile( q, "SELECT people_1.person_id AS people_1_person_id, " "people_1.type AS people_1_type, " "managers_1.person_id AS managers_1_person_id, " "managers_1.status AS managers_1_status, " "managers_1.manager_name AS managers_1_manager_name, " "companies.company_id AS companies_company_id, " "companies.name AS companies_name " "FROM companies JOIN (people AS people_1 LEFT OUTER JOIN " "managers AS managers_1 ON people_1.person_id = " "managers_1.person_id) ON companies.company_id = " "people_1.company_id", ) def test_load_only_subclass_from_relationship(self): s = Session() from sqlalchemy import inspect inspect(Company).add_property("managers", relationship(Manager)) q = ( s.query(Company) .join(Company.managers) .options( contains_eager(Company.managers).load_only( "status", "manager_name" ) ) ) self.assert_compile( q, "SELECT companies.company_id AS companies_company_id, " "companies.name AS companies_name, " "managers.person_id AS managers_person_id, " "people.person_id AS people_person_id, " "people.type AS people_type, " "managers.status AS managers_status, " "managers.manager_name AS managers_manager_name " "FROM companies JOIN (people JOIN managers ON people.person_id = " "managers.person_id) ON companies.company_id = people.company_id", ) def test_defer_on_wildcard_subclass(self): # pretty much the same as load_only except doesn't # exclude the primary key s = Session() q = ( s.query(Manager) .order_by(Person.person_id) .options(defer(".*"), undefer("status")) ) self.assert_compile( q, "SELECT managers.status AS managers_status " "FROM people JOIN managers ON " "people.person_id = managers.person_id ORDER BY people.person_id", ) def test_load_only_subclass_of_type(self): s = Session() q = s.query(Company).options( joinedload(Company.employees.of_type(Manager)).load_only("status") ) self.assert_compile( q, "SELECT companies.company_id AS companies_company_id, " "companies.name AS companies_name, " "anon_1.people_person_id AS anon_1_people_person_id, " "anon_1.people_type AS anon_1_people_type, " "anon_1.managers_person_id AS anon_1_managers_person_id, " "anon_1.managers_status AS anon_1_managers_status " "FROM companies LEFT OUTER JOIN " "(SELECT people.person_id AS people_person_id, " "people.company_id AS people_company_id, " "people.name AS people_name, people.type AS people_type, " "managers.person_id AS managers_person_id, " "managers.status AS managers_status, " "managers.manager_name AS managers_manager_name " "FROM people LEFT OUTER JOIN managers " "ON people.person_id = managers.person_id) AS anon_1 " "ON companies.company_id = anon_1.people_company_id " "ORDER BY anon_1.people_person_id", ) def test_wildcard_subclass_of_type(self): s = Session() q = s.query(Company).options( joinedload(Company.employees.of_type(Manager)).defer("*") ) self.assert_compile( q, "SELECT companies.company_id AS companies_company_id, " "companies.name AS companies_name " "FROM companies LEFT OUTER JOIN " "(SELECT people.person_id AS people_person_id, " "people.company_id AS people_company_id, " "people.name AS people_name, people.type AS people_type, " "managers.person_id AS managers_person_id, " "managers.status AS managers_status, " "managers.manager_name AS managers_manager_name " "FROM people LEFT OUTER JOIN managers " "ON people.person_id = managers.person_id) AS anon_1 " "ON companies.company_id = anon_1.people_company_id " "ORDER BY anon_1.people_person_id", ) def test_defer_super_name_on_subclass(self): s = Session() q = s.query(Manager).order_by(Person.person_id).options(defer("name")) self.assert_compile( q, "SELECT managers.person_id AS managers_person_id, " "people.person_id AS people_person_id, " "people.company_id AS people_company_id, " "people.type AS people_type, managers.status AS managers_status, " "managers.manager_name AS managers_manager_name " "FROM people JOIN managers " "ON people.person_id = managers.person_id " "ORDER BY people.person_id", ) class WithExpressionTest(fixtures.DeclarativeMappedTest): @classmethod def setup_classes(cls): Base = cls.DeclarativeBasic class A(fixtures.ComparableEntity, Base): __tablename__ = "a" id = Column(Integer, primary_key=True) x = Column(Integer) y = Column(Integer) my_expr = query_expression() bs = relationship("B", order_by="B.id") class B(fixtures.ComparableEntity, Base): __tablename__ = "b" id = Column(Integer, primary_key=True) a_id = Column(ForeignKey("a.id")) p = Column(Integer) q = Column(Integer) b_expr = query_expression() @classmethod def insert_data(cls): A, B = cls.classes("A", "B") s = Session() s.add_all( [ A(id=1, x=1, y=2, bs=[B(id=1, p=1, q=2), B(id=2, p=4, q=8)]), A(id=2, x=2, y=3), A(id=3, x=5, y=10, bs=[B(id=3, p=5, q=0)]), A(id=4, x=2, y=10, bs=[B(id=4, p=19, q=8), B(id=5, p=5, q=5)]), ] ) s.commit() def test_simple_expr(self): A = self.classes.A s = Session() a1 = ( s.query(A) .options(with_expression(A.my_expr, A.x + A.y)) .filter(A.x > 1) .order_by(A.id) ) eq_(a1.all(), [A(my_expr=5), A(my_expr=15), A(my_expr=12)]) def test_reuse_expr(self): A = self.classes.A s = Session() # so people will obv. want to say, "filter(A.my_expr > 10)". # but that means Query or Core has to post-modify the statement # after construction. expr = A.x + A.y a1 = ( s.query(A) .options(with_expression(A.my_expr, expr)) .filter(expr > 10) .order_by(expr) ) eq_(a1.all(), [A(my_expr=12), A(my_expr=15)]) def test_in_joinedload(self): A, B = self.classes("A", "B") s = Session() q = ( s.query(A) .options(joinedload(A.bs).with_expression(B.b_expr, B.p * A.x)) .filter(A.id.in_([3, 4])) .order_by(A.id) ) eq_( q.all(), [A(bs=[B(b_expr=25)]), A(bs=[B(b_expr=38), B(b_expr=10)])] ) def test_no_sql_not_set_up(self): A = self.classes.A s = Session() a1 = s.query(A).first() def go(): eq_(a1.my_expr, None) self.assert_sql_count(testing.db, go, 0) def test_dont_explode_on_expire_individual(self): A = self.classes.A s = Session() q = ( s.query(A) .options(with_expression(A.my_expr, A.x + A.y)) .filter(A.x > 1) .order_by(A.id) ) a1 = q.first() eq_(a1.my_expr, 5) s.expire(a1, ["my_expr"]) eq_(a1.my_expr, None) # comes back q = ( s.query(A) .options(with_expression(A.my_expr, A.x + A.y)) .filter(A.x > 1) .order_by(A.id) ) q.first() eq_(a1.my_expr, 5) def test_dont_explode_on_expire_whole(self): A = self.classes.A s = Session() q = ( s.query(A) .options(with_expression(A.my_expr, A.x + A.y)) .filter(A.x > 1) .order_by(A.id) ) a1 = q.first() eq_(a1.my_expr, 5) s.expire(a1) eq_(a1.my_expr, None) # comes back q = ( s.query(A) .options(with_expression(A.my_expr, A.x + A.y)) .filter(A.x > 1) .order_by(A.id) ) q.first() eq_(a1.my_expr, 5)