:ticket:`4662`
+.. _change_4994:
+
+Persistence-related cascade operations disallowed with viewonly=True
+---------------------------------------------------------------------
+
+When a :func:`.relationship` is set as ``viewonly=True`` using the
+:paramref:`.relationship.viewonly` flag, it indicates this relationship should
+only be used to load data from the database, and should not be mutated
+or involved in a persistence operation. In order to ensure this contract
+works successfully, the relationship can no longer specify
+:paramref:`.relationship.cascade` settings that make no sense in terms of
+"viewonly".
+
+The primary targets here are the "delete, delete-orphan" cascades, which
+through 1.3 continued to impact persistence even if viewonly were True, which
+is a bug; even if viewonly were True, an object would still cascade these
+two operations onto the related object if the parent were deleted or the
+object were detached. Rather than modify the cascade operations to check
+for viewonly, the configuration of both of these together is simply
+disallowed::
+
+ class User(Base):
+ # ...
+
+ # this is now an error
+ addresses = relationship(
+ "Address", viewonly=True, cascade="all, delete-orphan")
+
+The above will raise::
+
+ sqlalchemy.exc.ArgumentError: Cascade settings
+ "delete, delete-orphan, merge, save-update" apply to persistence
+ operations and should not be combined with a viewonly=True relationship.
+
+Applications that have this issue should be emitting a warning as of
+SQLAlchemy 1.3.12, and for the above error the solution is to remove
+the cascade settings for a viewonly relationship.
+
+
+:ticket:`4993`
+:ticket:`4994`
+
+
Behavior Changes - Core
========================
--- /dev/null
+.. change::
+ :tags: bug, orm
+ :tickets: 4994
+
+ An error is raised if any persistence-related "cascade" settings are made
+ on a :func:`.relationship` that also sets up viewonly=True. The "cascade"
+ settings now default to non-persistence related settings only when viewonly
+ is also set. This is the continuation from :ticket:`4993` where this
+ setting was changed to emit a warning in 1.3.
+
+ .. seealso::
+
+ :ref:`change_4994`
+
+
if cascade is not False:
self.cascade = cascade
+ elif self.viewonly:
+ self.cascade = "none"
else:
- self._set_cascade("save-update, merge", warn=False)
+ self.cascade = "save-update, merge"
self.order_by = order_by
def cascade(self, cascade):
self._set_cascade(cascade)
- def _set_cascade(self, cascade, warn=True):
+ def _set_cascade(self, cascade):
cascade = CascadeOptions(cascade)
- if warn and self.viewonly:
+ if self.viewonly:
non_viewonly = set(cascade).difference(
CascadeOptions._viewonly_cascades
)
if non_viewonly:
- # we are warning here rather than warn deprecated as this
- # setting actively does the wrong thing and Python shows
- # regular warnings more aggressively than deprecation warnings
- # by default. There's no other guard against setting active
- # persistence cascades under viewonly=True so this will raise
- # in 1.4.
- util.warn(
- 'Cascade settings "%s" should not be combined with a '
- "viewonly=True relationship. This configuration will "
- "raise an error in version 1.4. Note that in versions "
- "prior to 1.4, "
- "these cascade settings may still produce a mutating "
- "effect even though this relationship is marked as "
- "viewonly=True." % (", ".join(sorted(non_viewonly)))
+ raise sa_exc.ArgumentError(
+ 'Cascade settings "%s" apply to persistence operations '
+ "and should not be combined with a viewonly=True "
+ "relationship." % (", ".join(sorted(non_viewonly)))
)
if "mapper" in self.__dict__:
from sqlalchemy.orm import configure_mappers
from sqlalchemy.orm import create_session
from sqlalchemy.orm import exc as orm_exc
+from sqlalchemy.orm import foreign
from sqlalchemy.orm import mapper
from sqlalchemy.orm import object_mapper
from sqlalchemy.orm import relationship
from sqlalchemy.testing import assert_raises_message
from sqlalchemy.testing import eq_
from sqlalchemy.testing import fixtures
+from sqlalchemy.testing import in_
+from sqlalchemy.testing import not_in_
from sqlalchemy.testing.schema import Column
from sqlalchemy.testing.schema import Table
from test.orm import _fixtures
s.commit()
eq_(s.query(Language).count(), 0)
+
+
+class ViewonlyFlagWarningTest(fixtures.MappedTest):
+ """test for #4993.
+
+ In 1.4, this moves to test/orm/test_cascade, deprecation warnings
+ become errors, will then be for #4994.
+
+ """
+
+ @classmethod
+ def define_tables(cls, metadata):
+ Table(
+ "users",
+ metadata,
+ Column("id", Integer, primary_key=True),
+ Column("name", String(30)),
+ )
+ Table(
+ "orders",
+ metadata,
+ Column("id", Integer, primary_key=True),
+ Column("user_id", Integer),
+ Column("description", String(30)),
+ )
+
+ @classmethod
+ def setup_classes(cls):
+ class User(cls.Comparable):
+ pass
+
+ class Order(cls.Comparable):
+ pass
+
+ @testing.combinations(
+ ({"delete"}, {"delete"}),
+ (
+ {"all, delete-orphan"},
+ {"delete", "delete-orphan", "merge", "save-update"},
+ ),
+ ({"save-update, expunge"}, {"save-update"}),
+ )
+ def test_write_cascades(self, setting, settings_that_warn):
+ Order = self.classes.Order
+
+ assert_raises_message(
+ sa_exc.ArgumentError,
+ 'Cascade settings "%s" apply to persistence '
+ "operations" % (", ".join(sorted(settings_that_warn))),
+ relationship,
+ Order,
+ primaryjoin=(
+ self.tables.users.c.id == foreign(self.tables.orders.c.user_id)
+ ),
+ cascade=", ".join(sorted(setting)),
+ viewonly=True,
+ )
+
+ def test_expunge_cascade(self):
+ User, Order, orders, users = (
+ self.classes.User,
+ self.classes.Order,
+ self.tables.orders,
+ self.tables.users,
+ )
+
+ mapper(Order, orders)
+ mapper(
+ User,
+ users,
+ properties={
+ "orders": relationship(
+ Order,
+ primaryjoin=(
+ self.tables.users.c.id
+ == foreign(self.tables.orders.c.user_id)
+ ),
+ cascade="expunge",
+ viewonly=True,
+ )
+ },
+ )
+
+ sess = Session()
+ u = User(id=1, name="jack")
+ sess.add(u)
+ sess.add_all(
+ [
+ Order(id=1, user_id=1, description="someorder"),
+ Order(id=2, user_id=1, description="someotherorder"),
+ ]
+ )
+ sess.commit()
+
+ u1 = sess.query(User).first()
+ orders = u1.orders
+ eq_(len(orders), 2)
+
+ in_(orders[0], sess)
+ in_(orders[1], sess)
+
+ sess.expunge(u1)
+
+ not_in_(orders[0], sess)
+ not_in_(orders[1], sess)
+
+ def test_default_none_cascade(self):
+ User, Order, orders, users = (
+ self.classes.User,
+ self.classes.Order,
+ self.tables.orders,
+ self.tables.users,
+ )
+
+ mapper(Order, orders)
+ mapper(
+ User,
+ users,
+ properties={
+ "orders": relationship(
+ Order,
+ primaryjoin=(
+ self.tables.users.c.id
+ == foreign(self.tables.orders.c.user_id)
+ ),
+ viewonly=True,
+ )
+ },
+ )
+
+ sess = Session()
+ u1 = User(id=1, name="jack")
+ sess.add(u1)
+
+ o1, o2 = (
+ Order(id=1, user_id=1, description="someorder"),
+ Order(id=2, user_id=1, description="someotherorder"),
+ )
+
+ u1.orders.append(o1)
+ u1.orders.append(o2)
+
+ not_in_(o1, sess)
+ not_in_(o2, sess)
+
+ def test_default_merge_cascade(self):
+ User, Order, orders, users = (
+ self.classes.User,
+ self.classes.Order,
+ self.tables.orders,
+ self.tables.users,
+ )
+
+ mapper(Order, orders)
+ mapper(
+ User,
+ users,
+ properties={
+ "orders": relationship(
+ Order,
+ primaryjoin=(
+ self.tables.users.c.id
+ == foreign(self.tables.orders.c.user_id)
+ ),
+ viewonly=True,
+ )
+ },
+ )
+
+ sess = Session()
+ u1 = User(id=1, name="jack")
+
+ o1, o2 = (
+ Order(id=1, user_id=1, description="someorder"),
+ Order(id=2, user_id=1, description="someotherorder"),
+ )
+
+ u1.orders.append(o1)
+ u1.orders.append(o2)
+
+ u1 = sess.merge(u1)
+
+ assert not u1.orders
+
+ def test_default_cascade(self):
+ User, Order, orders, users = (
+ self.classes.User,
+ self.classes.Order,
+ self.tables.orders,
+ self.tables.users,
+ )
+
+ mapper(Order, orders)
+ umapper = mapper(
+ User,
+ users,
+ properties={
+ "orders": relationship(
+ Order,
+ primaryjoin=(
+ self.tables.users.c.id
+ == foreign(self.tables.orders.c.user_id)
+ ),
+ viewonly=True,
+ )
+ },
+ )
+
+ eq_(umapper.attrs["orders"].cascade, set())
+
+ def test_write_cascade_disallowed_w_viewonly(self):
+
+ Order = self.classes.Order
+
+ assert_raises_message(
+ sa_exc.ArgumentError,
+ 'Cascade settings "delete, delete-orphan, merge, save-update" '
+ "apply to persistence operations",
+ relationship,
+ Order,
+ primaryjoin=(
+ self.tables.users.c.id == foreign(self.tables.orders.c.user_id)
+ ),
+ cascade="all, delete, delete-orphan",
+ viewonly=True,
+ )
from sqlalchemy.testing import AssertsCompiledSQL
from sqlalchemy.testing import eq_
from sqlalchemy.testing import fixtures
-from sqlalchemy.testing import in_
from sqlalchemy.testing import is_
from sqlalchemy.testing import is_true
-from sqlalchemy.testing import not_in_
from sqlalchemy.testing.schema import Column
from sqlalchemy.testing.schema import Table
from sqlalchemy.testing.util import gc_collect
kw[flag] = value
rel = relationship(Order, **kw)
- if flag == "cascade":
- eq_(set(rel.cascade), {"delete", "delete-orphan"})
- else:
- eq_(getattr(rel, flag), value)
-
- @testing.combinations(
- ({"delete"}, {"delete"}),
- (
- {"all, delete-orphan"},
- {"delete", "delete-orphan", "merge", "save-update"},
- ),
- ({"save-update, expunge"}, {"save-update"}),
- )
- def test_write_cascades(self, setting, settings_that_warn):
- Order = self.classes.Order
-
- with testing.expect_warnings(
- r"Cascade settings \"%s\" should not be combined"
- % (", ".join(sorted(settings_that_warn)))
- ):
- relationship(
- Order,
- primaryjoin=(
- self.tables.users.c.id
- == foreign(self.tables.orders.c.user_id)
- ),
- cascade=", ".join(sorted(setting)),
- viewonly=True,
- )
-
- def test_expunge_cascade(self):
- User, Order, orders, users = (
- self.classes.User,
- self.classes.Order,
- self.tables.orders,
- self.tables.users,
- )
-
- mapper(Order, orders)
- mapper(
- User,
- users,
- properties={
- "orders": relationship(
- Order,
- primaryjoin=(
- self.tables.users.c.id
- == foreign(self.tables.orders.c.user_id)
- ),
- cascade="expunge",
- viewonly=True,
- )
- },
- )
-
- sess = Session()
- u = User(id=1, name="jack")
- sess.add(u)
- sess.add_all(
- [
- Order(id=1, user_id=1, description="someorder"),
- Order(id=2, user_id=1, description="someotherorder"),
- ]
- )
- sess.commit()
-
- u1 = sess.query(User).first()
- orders = u1.orders
- eq_(len(orders), 2)
-
- in_(orders[0], sess)
- in_(orders[1], sess)
-
- sess.expunge(u1)
-
- not_in_(orders[0], sess)
- not_in_(orders[1], sess)
-
- def test_default_save_update_cascade(self):
- User, Order, orders, users = (
- self.classes.User,
- self.classes.Order,
- self.tables.orders,
- self.tables.users,
- )
-
- mapper(Order, orders)
- mapper(
- User,
- users,
- properties={
- "orders": relationship(
- Order,
- primaryjoin=(
- self.tables.users.c.id
- == foreign(self.tables.orders.c.user_id)
- ),
- viewonly=True,
- )
- },
- )
-
- sess = Session()
- u1 = User(id=1, name="jack")
- sess.add(u1)
-
- o1, o2 = (
- Order(id=1, user_id=1, description="someorder"),
- Order(id=2, user_id=1, description="someotherorder"),
- )
-
- u1.orders.append(o1)
- u1.orders.append(o2)
-
- # in 1.4, this becomes "not_in_"
- in_(o1, sess)
- in_(o2, sess)
-
- def test_default_merge_cascade(self):
- User, Order, orders, users = (
- self.classes.User,
- self.classes.Order,
- self.tables.orders,
- self.tables.users,
- )
-
- mapper(Order, orders)
- mapper(
- User,
- users,
- properties={
- "orders": relationship(
- Order,
- primaryjoin=(
- self.tables.users.c.id
- == foreign(self.tables.orders.c.user_id)
- ),
- viewonly=True,
- )
- },
- )
-
- sess = Session()
- u1 = User(id=1, name="jack")
-
- o1, o2 = (
- Order(id=1, user_id=1, description="someorder"),
- Order(id=2, user_id=1, description="someotherorder"),
- )
-
- u1.orders.append(o1)
- u1.orders.append(o2)
-
- u1 = sess.merge(u1)
-
- # in 1.4, this becomes "assert not u1.orders", merge does not occur
- o1, o2 = u1.orders
-
- def test_default_cascade_didnt_change_yet(self):
- User, Order, orders, users = (
- self.classes.User,
- self.classes.Order,
- self.tables.orders,
- self.tables.users,
- )
-
- mapper(Order, orders)
- umapper = mapper(
- User,
- users,
- properties={
- "orders": relationship(
- Order,
- primaryjoin=(
- self.tables.users.c.id
- == foreign(self.tables.orders.c.user_id)
- ),
- viewonly=True,
- )
- },
- )
-
- # in 1.4 this becomes {}
- eq_(umapper.attrs["orders"].cascade, {"save-update", "merge"})
-
- def test_write_cascade_still_works_w_viewonly(self):
- """should be no longer possible in 1.4"""
-
- User, Order, orders, users = (
- self.classes.User,
- self.classes.Order,
- self.tables.orders,
- self.tables.users,
- )
-
- mapper(Order, orders)
- with testing.expect_warnings(r"Cascade settings"):
- mapper(
- User,
- users,
- properties={
- "orders": relationship(
- Order,
- primaryjoin=(
- self.tables.users.c.id
- == foreign(self.tables.orders.c.user_id)
- ),
- cascade="all, delete, delete-orphan",
- viewonly=True,
- )
- },
- )
-
- sess = Session()
- u = User(id=1, name="jack")
- sess.add(u)
- sess.add_all(
- [
- Order(id=1, user_id=1, description="someorder"),
- Order(id=2, user_id=1, description="someotherorder"),
- ]
- )
- sess.commit()
- eq_(sess.query(Order).count(), 2)
-
- sess.delete(u)
- sess.commit()
-
- eq_(sess.query(Order).count(), 0)
+ eq_(getattr(rel, flag), value)
class NonPrimaryMapperTest(_fixtures.FixtureTest, AssertsCompiledSQL):
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
