from sqlalchemy.orm import defer
from sqlalchemy.orm import exc as orm_exc
from sqlalchemy.orm import joinedload
-from sqlalchemy.orm import lazyload
from sqlalchemy.orm import Load
from sqlalchemy.orm import load_only
from sqlalchemy.orm import mapper
joinedload(User.orders).options,
Load(Order).joinedload(Order.items),
)
-
-
-class PathedCacheKeyTest(PathTest, QueryTest):
-
- run_create_tables = False
- run_inserts = None
- run_deletes = None
-
- def test_unbound_cache_key_included_safe(self):
- User, Address, Order, Item, SubItem = self.classes(
- "User", "Address", "Order", "Item", "SubItem"
- )
-
- query_path = self._make_path_registry([User, "orders"])
-
- opt = joinedload(User.orders).joinedload(Order.items)
- eq_(
- opt._generate_path_cache_key(query_path),
- (((Order, "items", Item, ("lazy", "joined")),)),
- )
-
- def test_unbound_cache_key_included_safe_multipath(self):
- User, Address, Order, Item, SubItem = self.classes(
- "User", "Address", "Order", "Item", "SubItem"
- )
-
- query_path = self._make_path_registry([User, "orders"])
-
- base = joinedload(User.orders)
- opt1 = base.joinedload(Order.items)
- opt2 = base.joinedload(Order.address)
-
- eq_(
- opt1._generate_path_cache_key(query_path),
- (((Order, "items", Item, ("lazy", "joined")),)),
- )
-
- eq_(
- opt2._generate_path_cache_key(query_path),
- (((Order, "address", Address, ("lazy", "joined")),)),
- )
-
- def test_bound_cache_key_included_safe_multipath(self):
- User, Address, Order, Item, SubItem = self.classes(
- "User", "Address", "Order", "Item", "SubItem"
- )
-
- query_path = self._make_path_registry([User, "orders"])
-
- base = Load(User).joinedload(User.orders)
- opt1 = base.joinedload(Order.items)
- opt2 = base.joinedload(Order.address)
-
- eq_(
- opt1._generate_path_cache_key(query_path),
- (((Order, "items", Item, ("lazy", "joined")),)),
- )
-
- eq_(
- opt2._generate_path_cache_key(query_path),
- (((Order, "address", Address, ("lazy", "joined")),)),
- )
-
- def test_bound_cache_key_included_safe(self):
- User, Address, Order, Item, SubItem = self.classes(
- "User", "Address", "Order", "Item", "SubItem"
- )
-
- query_path = self._make_path_registry([User, "orders"])
-
- opt = Load(User).joinedload(User.orders).joinedload(Order.items)
- eq_(
- opt._generate_path_cache_key(query_path),
- (((Order, "items", Item, ("lazy", "joined")),)),
- )
-
- def test_unbound_cache_key_excluded_on_other(self):
- User, Address, Order, Item, SubItem = self.classes(
- "User", "Address", "Order", "Item", "SubItem"
- )
-
- query_path = self._make_path_registry([User, "addresses"])
-
- opt = joinedload(User.orders).joinedload(Order.items)
- eq_(opt._generate_path_cache_key(query_path), None)
-
- def test_bound_cache_key_excluded_on_other(self):
- User, Address, Order, Item, SubItem = self.classes(
- "User", "Address", "Order", "Item", "SubItem"
- )
-
- query_path = self._make_path_registry([User, "addresses"])
-
- opt = Load(User).joinedload(User.orders).joinedload(Order.items)
- eq_(opt._generate_path_cache_key(query_path), None)
-
- def test_unbound_cache_key_excluded_on_aliased(self):
- User, Address, Order, Item, SubItem = self.classes(
- "User", "Address", "Order", "Item", "SubItem"
- )
-
- # query of:
- #
- # query(User).options(
- # joinedload(aliased(User).orders).joinedload(Order.items))
- #
- # we are lazy loading Order objects from User.orders
- # the path excludes our option so cache key should
- # be None
-
- query_path = self._make_path_registry([User, "orders"])
-
- opt = joinedload(aliased(User).orders).joinedload(Order.items)
- eq_(opt._generate_path_cache_key(query_path), None)
-
- def test_bound_cache_key_wildcard_one(self):
- # do not change this test, it is testing
- # a specific condition in Load._chop_path().
- User, Address = self.classes("User", "Address")
-
- query_path = self._make_path_registry([User, "addresses"])
-
- opt = Load(User).lazyload("*")
- eq_(opt._generate_path_cache_key(query_path), None)
-
- def test_unbound_cache_key_wildcard_one(self):
- User, Address = self.classes("User", "Address")
-
- query_path = self._make_path_registry([User, "addresses"])
-
- opt = lazyload("*")
- eq_(
- opt._generate_path_cache_key(query_path),
- (("relationship:_sa_default", ("lazy", "select")),),
- )
-
- def test_bound_cache_key_wildcard_two(self):
- User, Address, Order, Item, SubItem, Keyword = self.classes(
- "User", "Address", "Order", "Item", "SubItem", "Keyword"
- )
-
- query_path = self._make_path_registry([User])
-
- opt = Load(User).lazyload("orders").lazyload("*")
- eq_(
- opt._generate_path_cache_key(query_path),
- (
- ("orders", Order, ("lazy", "select")),
- ("orders", Order, "relationship:*", ("lazy", "select")),
- ),
- )
-
- def test_unbound_cache_key_wildcard_two(self):
- User, Address, Order, Item, SubItem, Keyword = self.classes(
- "User", "Address", "Order", "Item", "SubItem", "Keyword"
- )
-
- query_path = self._make_path_registry([User])
-
- opt = lazyload("orders").lazyload("*")
- eq_(
- opt._generate_path_cache_key(query_path),
- (
- ("orders", Order, ("lazy", "select")),
- ("orders", Order, "relationship:*", ("lazy", "select")),
- ),
- )
-
- def test_unbound_cache_key_of_type_subclass_relationship(self):
- User, Address, Order, Item, SubItem, Keyword = self.classes(
- "User", "Address", "Order", "Item", "SubItem", "Keyword"
- )
-
- query_path = self._make_path_registry([Order, "items", Item])
-
- opt = subqueryload(Order.items.of_type(SubItem)).subqueryload(
- SubItem.extra_keywords
- )
-
- eq_(
- opt._generate_path_cache_key(query_path),
- (
- (SubItem, ("lazy", "subquery")),
- ("extra_keywords", Keyword, ("lazy", "subquery")),
- ),
- )
-
- def test_unbound_cache_key_of_type_subclass_relationship_stringattr(self):
- User, Address, Order, Item, SubItem, Keyword = self.classes(
- "User", "Address", "Order", "Item", "SubItem", "Keyword"
- )
-
- query_path = self._make_path_registry([Order, "items", Item])
-
- opt = subqueryload(Order.items.of_type(SubItem)).subqueryload(
- "extra_keywords"
- )
-
- eq_(
- opt._generate_path_cache_key(query_path),
- (
- (SubItem, ("lazy", "subquery")),
- ("extra_keywords", Keyword, ("lazy", "subquery")),
- ),
- )
-
- def test_bound_cache_key_of_type_subclass_relationship(self):
- User, Address, Order, Item, SubItem, Keyword = self.classes(
- "User", "Address", "Order", "Item", "SubItem", "Keyword"
- )
-
- query_path = self._make_path_registry([Order, "items", Item])
-
- opt = (
- Load(Order)
- .subqueryload(Order.items.of_type(SubItem))
- .subqueryload(SubItem.extra_keywords)
- )
-
- eq_(
- opt._generate_path_cache_key(query_path),
- (
- (SubItem, ("lazy", "subquery")),
- ("extra_keywords", Keyword, ("lazy", "subquery")),
- ),
- )
-
- def test_bound_cache_key_of_type_subclass_string_relationship(self):
- User, Address, Order, Item, SubItem, Keyword = self.classes(
- "User", "Address", "Order", "Item", "SubItem", "Keyword"
- )
-
- query_path = self._make_path_registry([Order, "items", Item])
-
- opt = (
- Load(Order)
- .subqueryload(Order.items.of_type(SubItem))
- .subqueryload("extra_keywords")
- )
-
- eq_(
- opt._generate_path_cache_key(query_path),
- (
- (SubItem, ("lazy", "subquery")),
- ("extra_keywords", Keyword, ("lazy", "subquery")),
- ),
- )
-
- def test_unbound_cache_key_excluded_of_type_safe(self):
- User, Address, Order, Item, SubItem = self.classes(
- "User", "Address", "Order", "Item", "SubItem"
- )
- # query of:
- #
- # query(User).options(
- # subqueryload(User.orders).
- # subqueryload(Order.items.of_type(SubItem)))
- #
- #
- # we are lazy loading Address objects from User.addresses
- # the path excludes our option so cache key should
- # be None
-
- query_path = self._make_path_registry([User, "addresses"])
-
- opt = subqueryload(User.orders).subqueryload(
- Order.items.of_type(SubItem)
- )
- eq_(opt._generate_path_cache_key(query_path), None)
-
- def test_unbound_cache_key_excluded_of_type_unsafe(self):
- User, Address, Order, Item, SubItem = self.classes(
- "User", "Address", "Order", "Item", "SubItem"
- )
- # query of:
- #
- # query(User).options(
- # subqueryload(User.orders).
- # subqueryload(Order.items.of_type(aliased(SubItem))))
- #
- #
- # we are lazy loading Address objects from User.addresses
- # the path excludes our option so cache key should
- # be None
-
- query_path = self._make_path_registry([User, "addresses"])
-
- opt = subqueryload(User.orders).subqueryload(
- Order.items.of_type(aliased(SubItem))
- )
- eq_(opt._generate_path_cache_key(query_path), None)
-
- def test_bound_cache_key_excluded_of_type_safe(self):
- User, Address, Order, Item, SubItem = self.classes(
- "User", "Address", "Order", "Item", "SubItem"
- )
- # query of:
- #
- # query(User).options(
- # subqueryload(User.orders).
- # subqueryload(Order.items.of_type(SubItem)))
- #
- #
- # we are lazy loading Address objects from User.addresses
- # the path excludes our option so cache key should
- # be None
-
- query_path = self._make_path_registry([User, "addresses"])
-
- opt = (
- Load(User)
- .subqueryload(User.orders)
- .subqueryload(Order.items.of_type(SubItem))
- )
- eq_(opt._generate_path_cache_key(query_path), None)
-
- def test_bound_cache_key_excluded_of_type_unsafe(self):
- User, Address, Order, Item, SubItem = self.classes(
- "User", "Address", "Order", "Item", "SubItem"
- )
- # query of:
- #
- # query(User).options(
- # subqueryload(User.orders).
- # subqueryload(Order.items.of_type(aliased(SubItem))))
- #
- #
- # we are lazy loading Address objects from User.addresses
- # the path excludes our option so cache key should
- # be None
-
- query_path = self._make_path_registry([User, "addresses"])
-
- opt = (
- Load(User)
- .subqueryload(User.orders)
- .subqueryload(Order.items.of_type(aliased(SubItem)))
- )
- eq_(opt._generate_path_cache_key(query_path), None)
-
- def test_unbound_cache_key_included_of_type_safe(self):
- User, Address, Order, Item, SubItem = self.classes(
- "User", "Address", "Order", "Item", "SubItem"
- )
-
- query_path = self._make_path_registry([User, "orders"])
-
- opt = joinedload(User.orders).joinedload(Order.items.of_type(SubItem))
- eq_(
- opt._generate_path_cache_key(query_path),
- ((Order, "items", SubItem, ("lazy", "joined")),),
- )
-
- def test_bound_cache_key_included_of_type_safe(self):
- User, Address, Order, Item, SubItem = self.classes(
- "User", "Address", "Order", "Item", "SubItem"
- )
-
- query_path = self._make_path_registry([User, "orders"])
-
- opt = (
- Load(User)
- .joinedload(User.orders)
- .joinedload(Order.items.of_type(SubItem))
- )
-
- eq_(
- opt._generate_path_cache_key(query_path),
- ((Order, "items", SubItem, ("lazy", "joined")),),
- )
-
- def test_unbound_cache_key_included_unsafe_option_one(self):
- User, Address, Order, Item, SubItem = self.classes(
- "User", "Address", "Order", "Item", "SubItem"
- )
-
- query_path = self._make_path_registry([User, "orders"])
-
- opt = joinedload(User.orders).joinedload(
- Order.items.of_type(aliased(SubItem))
- )
- eq_(opt._generate_path_cache_key(query_path), False)
-
- def test_unbound_cache_key_included_unsafe_option_two(self):
- User, Address, Order, Item, SubItem = self.classes(
- "User", "Address", "Order", "Item", "SubItem"
- )
-
- query_path = self._make_path_registry([User, "orders", Order])
-
- opt = joinedload(User.orders).joinedload(
- Order.items.of_type(aliased(SubItem))
- )
- eq_(opt._generate_path_cache_key(query_path), False)
-
- def test_unbound_cache_key_included_unsafe_option_three(self):
- User, Address, Order, Item, SubItem = self.classes(
- "User", "Address", "Order", "Item", "SubItem"
- )
-
- query_path = self._make_path_registry([User, "orders", Order, "items"])
-
- opt = joinedload(User.orders).joinedload(
- Order.items.of_type(aliased(SubItem))
- )
- eq_(opt._generate_path_cache_key(query_path), False)
-
- def test_unbound_cache_key_included_unsafe_query(self):
- User, Address, Order, Item, SubItem = self.classes(
- "User", "Address", "Order", "Item", "SubItem"
- )
-
- au = aliased(User)
- query_path = self._make_path_registry([inspect(au), "orders"])
-
- opt = joinedload(au.orders).joinedload(Order.items)
- eq_(opt._generate_path_cache_key(query_path), False)
-
- def test_unbound_cache_key_included_safe_w_deferred(self):
- User, Address, Order, Item, SubItem = self.classes(
- "User", "Address", "Order", "Item", "SubItem"
- )
-
- query_path = self._make_path_registry([User, "addresses"])
-
- opt = (
- joinedload(User.addresses)
- .defer(Address.email_address)
- .defer(Address.user_id)
- )
- eq_(
- opt._generate_path_cache_key(query_path),
- (
- (
- Address,
- "email_address",
- ("deferred", True),
- ("instrument", True),
- ),
- (Address, "user_id", ("deferred", True), ("instrument", True)),
- ),
- )
-
- def test_unbound_cache_key_included_safe_w_deferred_multipath(self):
- User, Address, Order, Item, SubItem = self.classes(
- "User", "Address", "Order", "Item", "SubItem"
- )
-
- query_path = self._make_path_registry([User, "orders"])
-
- base = joinedload(User.orders)
- opt1 = base.joinedload(Order.items)
- opt2 = (
- base.joinedload(Order.address)
- .defer(Address.email_address)
- .defer(Address.user_id)
- )
-
- eq_(
- opt1._generate_path_cache_key(query_path),
- ((Order, "items", Item, ("lazy", "joined")),),
- )
-
- eq_(
- opt2._generate_path_cache_key(query_path),
- (
- (Order, "address", Address, ("lazy", "joined")),
- (
- Order,
- "address",
- Address,
- "email_address",
- ("deferred", True),
- ("instrument", True),
- ),
- (
- Order,
- "address",
- Address,
- "user_id",
- ("deferred", True),
- ("instrument", True),
- ),
- ),
- )
-
- def test_bound_cache_key_included_safe_w_deferred(self):
- User, Address, Order, Item, SubItem = self.classes(
- "User", "Address", "Order", "Item", "SubItem"
- )
-
- query_path = self._make_path_registry([User, "addresses"])
-
- opt = (
- Load(User)
- .joinedload(User.addresses)
- .defer(Address.email_address)
- .defer(Address.user_id)
- )
- eq_(
- opt._generate_path_cache_key(query_path),
- (
- (
- Address,
- "email_address",
- ("deferred", True),
- ("instrument", True),
- ),
- (Address, "user_id", ("deferred", True), ("instrument", True)),
- ),
- )
-
- def test_bound_cache_key_included_safe_w_deferred_multipath(self):
- User, Address, Order, Item, SubItem = self.classes(
- "User", "Address", "Order", "Item", "SubItem"
- )
-
- query_path = self._make_path_registry([User, "orders"])
-
- base = Load(User).joinedload(User.orders)
- opt1 = base.joinedload(Order.items)
- opt2 = (
- base.joinedload(Order.address)
- .defer(Address.email_address)
- .defer(Address.user_id)
- )
-
- eq_(
- opt1._generate_path_cache_key(query_path),
- ((Order, "items", Item, ("lazy", "joined")),),
- )
-
- eq_(
- opt2._generate_path_cache_key(query_path),
- (
- (Order, "address", Address, ("lazy", "joined")),
- (
- Order,
- "address",
- Address,
- "email_address",
- ("deferred", True),
- ("instrument", True),
- ),
- (
- Order,
- "address",
- Address,
- "user_id",
- ("deferred", True),
- ("instrument", True),
- ),
- ),
- )
-
- def test_unbound_cache_key_included_safe_w_option(self):
- User, Address, Order, Item, SubItem = self.classes(
- "User", "Address", "Order", "Item", "SubItem"
- )
-
- opt = (
- defaultload("orders")
- .joinedload("items", innerjoin=True)
- .defer("description")
- )
- query_path = self._make_path_registry([User, "orders"])
-
- eq_(
- opt._generate_path_cache_key(query_path),
- (
- (
- Order,
- "items",
- Item,
- ("lazy", "joined"),
- ("innerjoin", True),
- ),
- (
- Order,
- "items",
- Item,
- "description",
- ("deferred", True),
- ("instrument", True),
- ),
- ),
- )
-
- def test_bound_cache_key_excluded_on_aliased(self):
- User, Address, Order, Item, SubItem = self.classes(
- "User", "Address", "Order", "Item", "SubItem"
- )
-
- query_path = self._make_path_registry([User, "orders"])
-
- au = aliased(User)
- opt = Load(au).joinedload(au.orders).joinedload(Order.items)
- eq_(opt._generate_path_cache_key(query_path), None)
-
- def test_bound_cache_key_included_unsafe_option_one(self):
- User, Address, Order, Item, SubItem = self.classes(
- "User", "Address", "Order", "Item", "SubItem"
- )
-
- query_path = self._make_path_registry([User, "orders"])
-
- opt = (
- Load(User)
- .joinedload(User.orders)
- .joinedload(Order.items.of_type(aliased(SubItem)))
- )
- eq_(opt._generate_path_cache_key(query_path), False)
-
- def test_bound_cache_key_included_unsafe_option_two(self):
- User, Address, Order, Item, SubItem = self.classes(
- "User", "Address", "Order", "Item", "SubItem"
- )
-
- query_path = self._make_path_registry([User, "orders", Order])
-
- opt = (
- Load(User)
- .joinedload(User.orders)
- .joinedload(Order.items.of_type(aliased(SubItem)))
- )
- eq_(opt._generate_path_cache_key(query_path), False)
-
- def test_bound_cache_key_included_unsafe_option_three(self):
- User, Address, Order, Item, SubItem = self.classes(
- "User", "Address", "Order", "Item", "SubItem"
- )
-
- query_path = self._make_path_registry([User, "orders", Order, "items"])
-
- opt = (
- Load(User)
- .joinedload(User.orders)
- .joinedload(Order.items.of_type(aliased(SubItem)))
- )
- eq_(opt._generate_path_cache_key(query_path), False)
-
- def test_bound_cache_key_included_unsafe_query(self):
- User, Address, Order, Item, SubItem = self.classes(
- "User", "Address", "Order", "Item", "SubItem"
- )
-
- au = aliased(User)
- query_path = self._make_path_registry([inspect(au), "orders"])
-
- opt = Load(au).joinedload(au.orders).joinedload(Order.items)
- eq_(opt._generate_path_cache_key(query_path), False)
-
- def test_bound_cache_key_included_safe_w_option(self):
- User, Address, Order, Item, SubItem = self.classes(
- "User", "Address", "Order", "Item", "SubItem"
- )
-
- opt = (
- Load(User)
- .defaultload("orders")
- .joinedload("items", innerjoin=True)
- .defer("description")
- )
- query_path = self._make_path_registry([User, "orders"])
-
- eq_(
- opt._generate_path_cache_key(query_path),
- (
- (
- Order,
- "items",
- Item,
- ("lazy", "joined"),
- ("innerjoin", True),
- ),
- (
- Order,
- "items",
- Item,
- "description",
- ("deferred", True),
- ("instrument", True),
- ),
- ),
- )
-
- def test_unbound_cache_key_included_safe_w_loadonly_strs(self):
- User, Address, Order, Item, SubItem = self.classes(
- "User", "Address", "Order", "Item", "SubItem"
- )
-
- query_path = self._make_path_registry([User, "addresses"])
-
- opt = defaultload(User.addresses).load_only("id", "email_address")
- eq_(
- opt._generate_path_cache_key(query_path),
- (
- (Address, "id", ("deferred", False), ("instrument", True)),
- (
- Address,
- "email_address",
- ("deferred", False),
- ("instrument", True),
- ),
- (
- Address,
- "column:*",
- ("deferred", True),
- ("instrument", True),
- ("undefer_pks", True),
- ),
- ),
- )
-
- def test_unbound_cache_key_included_safe_w_loadonly_props(self):
- User, Address, Order, Item, SubItem = self.classes(
- "User", "Address", "Order", "Item", "SubItem"
- )
-
- query_path = self._make_path_registry([User, "addresses"])
-
- opt = defaultload(User.addresses).load_only(
- Address.id, Address.email_address
- )
- eq_(
- opt._generate_path_cache_key(query_path),
- (
- (Address, "id", ("deferred", False), ("instrument", True)),
- (
- Address,
- "email_address",
- ("deferred", False),
- ("instrument", True),
- ),
- (
- Address,
- "column:*",
- ("deferred", True),
- ("instrument", True),
- ("undefer_pks", True),
- ),
- ),
- )
-
- def test_bound_cache_key_included_safe_w_loadonly(self):
- User, Address, Order, Item, SubItem = self.classes(
- "User", "Address", "Order", "Item", "SubItem"
- )
-
- query_path = self._make_path_registry([User, "addresses"])
-
- opt = (
- Load(User)
- .defaultload(User.addresses)
- .load_only("id", "email_address")
- )
- eq_(
- opt._generate_path_cache_key(query_path),
- (
- (Address, "id", ("deferred", False), ("instrument", True)),
- (
- Address,
- "email_address",
- ("deferred", False),
- ("instrument", True),
- ),
- (
- Address,
- "column:*",
- ("deferred", True),
- ("instrument", True),
- ("undefer_pks", True),
- ),
- ),
- )
-
- def test_unbound_cache_key_undefer_group(self):
- User, Address = self.classes("User", "Address")
-
- query_path = self._make_path_registry([User, "addresses"])
-
- opt = defaultload(User.addresses).undefer_group("xyz")
-
- eq_(
- opt._generate_path_cache_key(query_path),
- ((Address, "column:*", ("undefer_group_xyz", True)),),
- )
-
- def test_bound_cache_key_undefer_group(self):
- User, Address = self.classes("User", "Address")
-
- query_path = self._make_path_registry([User, "addresses"])
-
- opt = Load(User).defaultload(User.addresses).undefer_group("xyz")
-
- eq_(
- opt._generate_path_cache_key(query_path),
- ((Address, "column:*", ("undefer_group_xyz", True)),),
- )