Method _restore_snapshot

lib/sqlalchemy/orm/session.py:1098–1128

Restore the restoration state taken before a transaction began. Corresponds to a rollback.

(self, dirty_only: bool = False)

        self._key_switches = weakref.WeakKeyDictionary()

    def _restore_snapshot(self, dirty_only: bool = False) -> None:
        """Restore the restoration state taken before a transaction began.

        Corresponds to a rollback.

        """
        assert self._is_transaction_boundary

        to_expunge = set(self._new).union(self.session._new)
        self.session._expunge_states(to_expunge, to_transient=True)

        for s, (oldkey, newkey) in self._key_switches.items():
            # we probably can do this conditionally based on
            # if we expunged or not, but safe_discard does that anyway
            self.session.identity_map.safe_discard(s)

            # restore the old key
            s.key = oldkey

            # now restore the object, but only if we didn't expunge
            if s not in to_expunge:
                self.session.identity_map.replace(s)

        for s in set(self._deleted).union(self.session._deleted):
            self.session._update_impl(s, revert_deletion=True)

        assert not self.session._deleted

        for s in self.session.identity_map.all_states():
            if not dirty_only or s.modified or s in self._dirty:
                s._expire(s.dict, self.session.identity_map._modified)

    def _remove_snapshot(self) -> None:
        """Remove the restoration state taken before a transaction began.