from sqlalchemy.orm import with_loader_criteria
from sqlalchemy.orm import with_parent
from sqlalchemy.orm import with_polymorphic
+from sqlalchemy.orm.collections import attribute_mapped_collection
from sqlalchemy.orm.collections import collection
from sqlalchemy.orm.util import polymorphic_union
from sqlalchemy.sql import elements
"FROM addresses WHERE :param_2 = addresses.user_id) AS anon_1",
checkparams={"param_1": 7, "param_2": 8},
)
+
+
+class CollectionCascadesDespiteBackrefTest(fixtures.TestBase):
+ """test old cascade_backrefs behavior
+
+ see test/orm/test_cascade.py::class CollectionCascadesNoBackrefTest
+ for the future version
+
+ """
+
+ @testing.fixture
+ def cascade_fixture(self, registry):
+ def go(collection_class):
+ @registry.mapped
+ class A(object):
+ __tablename__ = "a"
+
+ id = Column(Integer, primary_key=True)
+ bs = relationship(
+ "B", backref="a", collection_class=collection_class
+ )
+
+ @registry.mapped
+ class B(object):
+ __tablename__ = "b_"
+ id = Column(Integer, primary_key=True)
+ a_id = Column(ForeignKey("a.id"))
+ key = Column(String)
+
+ return A, B
+
+ yield go
+
+ @testing.combinations(
+ (set, "add"),
+ (list, "append"),
+ (attribute_mapped_collection("key"), "__setitem__"),
+ (attribute_mapped_collection("key"), "setdefault"),
+ (attribute_mapped_collection("key"), "update_dict"),
+ (attribute_mapped_collection("key"), "update_kw"),
+ argnames="collection_class,methname",
+ )
+ @testing.combinations((True,), (False,), argnames="future")
+ def test_cascades_on_collection(
+ self, cascade_fixture, collection_class, methname, future
+ ):
+ A, B = cascade_fixture(collection_class)
+
+ s = Session(future=future)
+
+ a1 = A()
+ s.add(a1)
+
+ b1 = B(key="b1")
+ b2 = B(key="b2")
+ b3 = B(key="b3")
+
+ if future:
+ dep_ctx = util.nullcontext
+ else:
+
+ def dep_ctx():
+ return assertions.expect_deprecated_20(
+ '"B" object is being merged into a Session along the '
+ 'backref cascade path for relationship "A.bs"'
+ )
+
+ with dep_ctx():
+ b1.a = a1
+ with dep_ctx():
+ b3.a = a1
+
+ if future:
+ assert b1 not in s
+ assert b3 not in s
+ else:
+ assert b1 in s
+ assert b3 in s
+
+ if methname == "__setitem__":
+ meth = getattr(a1.bs, methname)
+ meth(b1.key, b1)
+ meth(b2.key, b2)
+ elif methname == "setdefault":
+ meth = getattr(a1.bs, methname)
+ meth(b1.key, b1)
+ meth(b2.key, b2)
+ elif methname == "update_dict" and isinstance(a1.bs, dict):
+ a1.bs.update({b1.key: b1, b2.key: b2})
+ elif methname == "update_kw" and isinstance(a1.bs, dict):
+ a1.bs.update(b1=b1, b2=b2)
+ else:
+ meth = getattr(a1.bs, methname)
+ meth(b1)
+ meth(b2)
+
+ assert b1 in s
+ assert b2 in s
+
+ # future version:
+ if future:
+ assert b3 not in s # the event never triggers from reverse
+ else:
+ # old behavior
+ assert b3 in s
+
+
+class LoadOnFKsTest(fixtures.DeclarativeMappedTest):
+ @classmethod
+ def setup_classes(cls):
+ Base = cls.DeclarativeBasic
+
+ class Parent(Base):
+ __tablename__ = "parent"
+ __table_args__ = {"mysql_engine": "InnoDB"}
+
+ id = Column(
+ Integer, primary_key=True, test_needs_autoincrement=True
+ )
+
+ class Child(Base):
+ __tablename__ = "child"
+ __table_args__ = {"mysql_engine": "InnoDB"}
+
+ id = Column(
+ Integer, primary_key=True, test_needs_autoincrement=True
+ )
+ parent_id = Column(Integer, ForeignKey("parent.id"))
+
+ parent = relationship(Parent, backref=backref("children"))
+
+ @testing.fixture
+ def parent_fixture(self, connection):
+ Parent, Child = self.classes("Parent", "Child")
+
+ sess = fixture_session(bind=connection, autoflush=False)
+ p1 = Parent()
+ p2 = Parent()
+ c1, c2 = Child(), Child()
+ c1.parent = p1
+ sess.add_all([p1, p2])
+ assert c1 in sess
+
+ yield sess, p1, p2, c1, c2
+
+ sess.close()
+
+ def test_enable_rel_loading_on_persistent_allows_backref_event(
+ self, parent_fixture
+ ):
+ sess, p1, p2, c1, c2 = parent_fixture
+ Parent, Child = self.classes("Parent", "Child")
+
+ c3 = Child()
+ sess.enable_relationship_loading(c3)
+ c3.parent_id = p1.id
+ with assertions.expect_deprecated_20(
+ '"Child" object is being merged into a Session along the '
+ 'backref cascade path for relationship "Parent.children"'
+ ):
+ c3.parent = p1
+
+ # backref fired off when c3.parent was set,
+ # because the "old" value was None
+ # change as of [ticket:3708]
+ assert c3 in p1.children
+
+ def test_enable_rel_loading_allows_backref_event(self, parent_fixture):
+ sess, p1, p2, c1, c2 = parent_fixture
+ Parent, Child = self.classes("Parent", "Child")
+
+ c3 = Child()
+ sess.enable_relationship_loading(c3)
+ c3.parent_id = p1.id
+
+ with assertions.expect_deprecated_20(
+ '"Child" object is being merged into a Session along the '
+ 'backref cascade path for relationship "Parent.children"'
+ ):
+ c3.parent = p1
+
+ # backref fired off when c3.parent was set,
+ # because the "old" value was None
+ # change as of [ticket:3708]
+ assert c3 in p1.children
+
+
+class LazyTest(_fixtures.FixtureTest):
+ run_inserts = "once"
+ run_deletes = None
+
+ def test_backrefs_dont_lazyload(self):
+ users, Address, addresses, User = (
+ self.tables.users,
+ self.classes.Address,
+ self.tables.addresses,
+ self.classes.User,
+ )
+
+ self.mapper_registry.map_imperatively(
+ User,
+ users,
+ properties={"addresses": relationship(Address, backref="user")},
+ )
+ self.mapper_registry.map_imperatively(Address, addresses)
+ sess = fixture_session(autoflush=False)
+ ad = sess.query(Address).filter_by(id=1).one()
+ assert ad.user.id == 7
+
+ def go():
+ ad.user = None
+ assert ad.user is None
+
+ self.assert_sql_count(testing.db, go, 0)
+
+ u1 = sess.query(User).filter_by(id=7).one()
+
+ def go():
+ assert ad not in u1.addresses
+
+ self.assert_sql_count(testing.db, go, 1)
+
+ sess.expire(u1, ["addresses"])
+
+ def go():
+ assert ad in u1.addresses
+
+ self.assert_sql_count(testing.db, go, 1)
+
+ sess.expire(u1, ["addresses"])
+ ad2 = Address()
+
+ def go():
+ with assertions.expect_deprecated_20(
+ ".* object is being merged into a Session along the "
+ "backref cascade path for relationship "
+ ):
+ ad2.user = u1
+ assert ad2.user is u1
+
+ self.assert_sql_count(testing.db, go, 0)
+
+ def go():
+ assert ad2 in u1.addresses
+
+ self.assert_sql_count(testing.db, go, 1)
from sqlalchemy.orm import backref
from sqlalchemy.orm import declarative_base
from sqlalchemy.orm import relationship
-from sqlalchemy.orm import Session
from sqlalchemy.orm.attributes import instance_state
from sqlalchemy.testing import AssertsExecutionResults
from sqlalchemy.testing import fixtures
self.assert_sql_count(testing.db, go, 0)
-class LoadOnFKsTest(AssertsExecutionResults, fixtures.TestBase):
- __leave_connections_for_teardown__ = True
-
- def setup_test(self):
- global Parent, Child, Base
- Base = declarative_base()
+class LoadOnFKsTest(fixtures.DeclarativeMappedTest):
+ @classmethod
+ def setup_classes(cls):
+ Base = cls.DeclarativeBasic
class Parent(Base):
__tablename__ = "parent"
parent = relationship(Parent, backref=backref("children"))
- Base.metadata.create_all(testing.db)
-
- global sess, p1, p2, c1, c2
- sess = Session(bind=testing.db)
+ @testing.fixture
+ def parent_fixture(self, connection):
+ Parent, Child = self.classes("Parent", "Child")
+ sess = fixture_session(bind=connection)
p1 = Parent()
p2 = Parent()
c1, c2 = Child(), Child()
sess.add_all([p1, p2])
assert c1 in sess
- sess.commit()
+ sess.flush()
- def teardown_test(self):
- sess.rollback()
- Base.metadata.drop_all(testing.db)
+ Child.parent.property.load_on_pending = False
- def test_load_on_pending_allows_backref_event(self):
- Child.parent.property.load_on_pending = True
- sess.autoflush = False
- c3 = Child()
- sess.add(c3)
- c3.parent_id = p1.id
- c3.parent = p1
+ sess.expire_all()
- # backref fired off when c3.parent was set,
- # because the "old" value was None.
- # change as of [ticket:3708]
- assert c3 in p1.children
+ yield sess, p1, p2, c1, c2
- def test_enable_rel_loading_allows_backref_event(self):
- sess.autoflush = False
- c3 = Child()
- sess.enable_relationship_loading(c3)
- c3.parent_id = p1.id
- c3.parent = p1
+ sess.close()
- # backref fired off when c3.parent was set,
- # because the "old" value was None
- # change as of [ticket:3708]
- assert c3 in p1.children
+ def test_m2o_history_on_persistent_allows_backref_event(
+ self, parent_fixture
+ ):
+ sess, p1, p2, c1, c2 = parent_fixture
+ Parent, Child = self.classes("Parent", "Child")
- def test_m2o_history_on_persistent_allows_backref_event(self):
c3 = Child()
sess.add(c3)
c3.parent_id = p1.id
assert c3 in p1.children
- def test_load_on_persistent_allows_backref_event(self):
+ def test_load_on_persistent_allows_backref_event(self, parent_fixture):
+ sess, p1, p2, c1, c2 = parent_fixture
+ Parent, Child = self.classes("Parent", "Child")
+
Child.parent.property.load_on_pending = True
c3 = Child()
sess.add(c3)
assert c3 in p1.children
- def test_enable_rel_loading_on_persistent_allows_backref_event(self):
+ def test_load_on_pending_allows_backref_event(self, parent_fixture):
+ sess, p1, p2, c1, c2 = parent_fixture
+ Parent, Child = self.classes("Parent", "Child")
+
+ sess.autoflush = False
+
+ Child.parent.property.load_on_pending = True
c3 = Child()
- sess.enable_relationship_loading(c3)
+ sess.add(c3)
c3.parent_id = p1.id
+
c3.parent = p1
# backref fired off when c3.parent was set,
- # because the "old" value was None
+ # because the "old" value was None.
# change as of [ticket:3708]
assert c3 in p1.children
- def test_no_load_on_pending_allows_backref_event(self):
+ def test_no_load_on_pending_allows_backref_event(self, parent_fixture):
+ sess, p1, p2, c1, c2 = parent_fixture
+ Parent, Child = self.classes("Parent", "Child")
+
# users who stick with the program and don't use
# 'load_on_pending' get expected behavior
assert c3 in p1.children
- def test_autoflush_on_pending(self):
+ def test_autoflush_on_pending(self, parent_fixture):
+ sess, p1, p2, c1, c2 = parent_fixture
+ Parent, Child = self.classes("Parent", "Child")
+
# ensure p1.id is not expired
p1.id
# pendings don't autoflush
assert c3.parent is None
- def test_autoflush_load_on_pending_on_pending(self):
+ def test_autoflush_load_on_pending_on_pending(self, parent_fixture):
+ sess, p1, p2, c1, c2 = parent_fixture
+ Parent, Child = self.classes("Parent", "Child")
+
# ensure p1.id is not expired
p1.id
# ...unless the flag is on
assert c3.parent is p1
- def test_collection_load_from_pending_populated(self):
+ def test_collection_load_from_pending_populated(self, parent_fixture):
+ sess, p1, p2, c1, c2 = parent_fixture
+ Parent, Child = self.classes("Parent", "Child")
+
Parent.children.property.load_on_pending = True
p2 = Parent(id=p1.id)
sess.add(p2)
self.assert_sql_count(testing.db, go, 1)
- def test_collection_load_from_pending_no_sql(self):
+ def test_collection_load_from_pending_no_sql(self, parent_fixture):
+ sess, p1, p2, c1, c2 = parent_fixture
+ Parent, Child = self.classes("Parent", "Child")
+
Parent.children.property.load_on_pending = True
p2 = Parent(id=None)
sess.add(p2)
self.assert_sql_count(testing.db, go, 0)
- def test_load_on_pending_with_set(self):
+ def test_load_on_pending_with_set(self, parent_fixture):
+ sess, p1, p2, c1, c2 = parent_fixture
+ Parent, Child = self.classes("Parent", "Child")
+
Child.parent.property.load_on_pending = True
p1.children
self.assert_sql_count(testing.db, go, 0)
- def test_backref_doesnt_double(self):
+ def test_backref_doesnt_double(self, parent_fixture):
+ sess, p1, p2, c1, c2 = parent_fixture
+ Parent, Child = self.classes("Parent", "Child")
+
Child.parent.property.load_on_pending = True
sess.autoflush = False
p1.children
c3.parent = p1
assert len(p1.children) == 2
- def test_m2o_lazy_loader_on_persistent(self):
+ @testing.combinations(True, False, argnames="loadfk")
+ @testing.combinations(True, False, argnames="loadrel")
+ @testing.combinations(True, False, argnames="autoflush")
+ @testing.combinations(True, False, argnames="manualflush")
+ @testing.combinations(True, False, argnames="fake_autoexpire")
+ def test_m2o_lazy_loader_on_persistent(
+ self,
+ parent_fixture,
+ loadfk,
+ loadrel,
+ autoflush,
+ manualflush,
+ fake_autoexpire,
+ ):
"""Compare the behaviors from the lazyloader using
the "committed" state in all cases, vs. the lazyloader
using the "current" state in all cases except during flush.
"""
- for loadfk in (True, False):
- for loadrel in (True, False):
- for autoflush in (True, False):
- for manualflush in (True, False):
- for fake_autoexpire in (True, False):
- sess.autoflush = autoflush
-
- if loadfk:
- c1.parent_id
- if loadrel:
- c1.parent
-
- c1.parent_id = p2.id
-
- if manualflush:
- sess.flush()
-
- # fake_autoexpire refers to the eventual
- # auto-expire of 'parent' when c1.parent_id
- # is altered.
- if fake_autoexpire:
- sess.expire(c1, ["parent"])
-
- # old 0.6 behavior
- # if manualflush and (not loadrel or
- # fake_autoexpire):
- # # a flush occurs, we get p2
- # assert c1.parent is p2
- # elif not loadrel and not loadfk:
- # # problematically - we get None since
- # # committed state
- # # is empty when c1.parent_id was mutated,
- # # since we want
- # # to save on selects. this is
- # # why the patch goes in in 0.6 - this is
- # # mostly a bug.
- # assert c1.parent is None
- # else:
- # # if things were loaded, autoflush doesn't
- # # even happen.
- # assert c1.parent is p1
-
- # new behavior
- if loadrel and not fake_autoexpire:
- assert c1.parent is p1
- else:
- assert c1.parent is p2
-
- sess.rollback()
-
- def test_m2o_lazy_loader_on_pending(self):
- for loadonpending in (False, True):
- for autoflush in (False, True):
- for manualflush in (False, True):
- Child.parent.property.load_on_pending = loadonpending
- sess.autoflush = autoflush
-
- # ensure p2.id not expired
- p2.id
-
- c2 = Child()
- sess.add(c2)
- c2.parent_id = p2.id
-
- if manualflush:
- sess.flush()
-
- if loadonpending or manualflush:
- assert c2.parent is p2
- else:
- assert c2.parent is None
-
- sess.rollback()
-
- def test_m2o_lazy_loader_on_transient(self):
- for loadonpending in (False, True):
- for attach in (False, True):
- for autoflush in (False, True):
- for manualflush in (False, True):
- for enable_relationship_rel in (False, True):
- Child.parent.property.load_on_pending = (
- loadonpending
- )
- sess.autoflush = autoflush
- c2 = Child()
-
- if attach:
- state = instance_state(c2)
- state.session_id = sess.hash_key
-
- if enable_relationship_rel:
- sess.enable_relationship_loading(c2)
-
- c2.parent_id = p2.id
-
- if manualflush:
- sess.flush()
-
- if (
- loadonpending and attach
- ) or enable_relationship_rel:
- assert c2.parent is p2
- else:
- assert c2.parent is None
-
- sess.rollback()
+ sess, p1, p2, c1, c2 = parent_fixture
+ Parent, Child = self.classes("Parent", "Child")
+
+ sess.autoflush = autoflush
+
+ if loadfk:
+ c1.parent_id
+ if loadrel:
+ c1.parent
+
+ c1.parent_id = p2.id
+
+ if manualflush:
+ sess.flush()
+
+ # fake_autoexpire refers to the eventual
+ # auto-expire of 'parent' when c1.parent_id
+ # is altered.
+ if fake_autoexpire:
+ sess.expire(c1, ["parent"])
+
+ # old 0.6 behavior
+ # if manualflush and (not loadrel or
+ # fake_autoexpire):
+ # # a flush occurs, we get p2
+ # assert c1.parent is p2
+ # elif not loadrel and not loadfk:
+ # # problematically - we get None since
+ # # committed state
+ # # is empty when c1.parent_id was mutated,
+ # # since we want
+ # # to save on selects. this is
+ # # why the patch goes in in 0.6 - this is
+ # # mostly a bug.
+ # assert c1.parent is None
+ # else:
+ # # if things were loaded, autoflush doesn't
+ # # even happen.
+ # assert c1.parent is p1
+
+ # new behavior
+ if loadrel and not fake_autoexpire:
+ assert c1.parent is p1
+ else:
+ assert c1.parent is p2
+
+ @testing.combinations(True, False, argnames="loadonpending")
+ @testing.combinations(True, False, argnames="autoflush")
+ @testing.combinations(True, False, argnames="manualflush")
+ def test_m2o_lazy_loader_on_pending(
+ self, parent_fixture, loadonpending, autoflush, manualflush
+ ):
+ sess, p1, p2, c1, c2 = parent_fixture
+ Parent, Child = self.classes("Parent", "Child")
+
+ Child.parent.property.load_on_pending = loadonpending
+ sess.autoflush = autoflush
+
+ # ensure p2.id not expired
+ p2.id
+
+ c2 = Child()
+ sess.add(c2)
+ c2.parent_id = p2.id
+
+ if manualflush:
+ sess.flush()
+
+ if loadonpending or manualflush:
+ assert c2.parent is p2
+ else:
+ assert c2.parent is None
+
+ @testing.combinations(True, False, argnames="loadonpending")
+ @testing.combinations(True, False, argnames="attach")
+ @testing.combinations(True, False, argnames="autoflush")
+ @testing.combinations(True, False, argnames="manualflush")
+ @testing.combinations(True, False, argnames="enable_relationship_rel")
+ def test_m2o_lazy_loader_on_transient(
+ self,
+ parent_fixture,
+ loadonpending,
+ attach,
+ autoflush,
+ manualflush,
+ enable_relationship_rel,
+ ):
+ sess, p1, p2, c1, c2 = parent_fixture
+ Parent, Child = self.classes("Parent", "Child")
+
+ Child.parent.property.load_on_pending = loadonpending
+ sess.autoflush = autoflush
+ c2 = Child()
+
+ if attach:
+ state = instance_state(c2)
+ state.session_id = sess.hash_key
+
+ if enable_relationship_rel:
+ sess.enable_relationship_loading(c2)
+
+ c2.parent_id = p2.id
+
+ if manualflush:
+ sess.flush()
+
+ if (loadonpending and attach) or enable_relationship_rel:
+ assert c2.parent is p2
+ else:
+ assert c2.parent is None
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
