(
self, state: InstanceState[Any], obj: object, head: bool
)
| 3666 | self._delete_impl(object_state(instance), instance, head=True) |
| 3667 | |
| 3668 | def _delete_impl( |
| 3669 | self, state: InstanceState[Any], obj: object, head: bool |
| 3670 | ) -> None: |
| 3671 | if state.key is None: |
| 3672 | if head: |
| 3673 | raise sa_exc.InvalidRequestError( |
| 3674 | "Instance '%s' is not persisted" % state_str(state) |
| 3675 | ) |
| 3676 | else: |
| 3677 | return |
| 3678 | |
| 3679 | to_attach = self._before_attach(state, obj) |
| 3680 | |
| 3681 | if state in self._deleted: |
| 3682 | return |
| 3683 | |
| 3684 | self.identity_map.add(state) |
| 3685 | |
| 3686 | if to_attach: |
| 3687 | self._after_attach(state, obj) |
| 3688 | |
| 3689 | if head: |
| 3690 | # grab the cascades before adding the item to the deleted list |
| 3691 | # so that autoflush does not delete the item |
| 3692 | # the strong reference to the instance itself is significant here |
| 3693 | cascade_states = list( |
| 3694 | state.manager.mapper.cascade_iterator("delete", state) |
| 3695 | ) |
| 3696 | else: |
| 3697 | cascade_states = None |
| 3698 | |
| 3699 | self._deleted[state] = obj |
| 3700 | |
| 3701 | if head: |
| 3702 | if TYPE_CHECKING: |
| 3703 | assert cascade_states is not None |
| 3704 | for o, m, st_, dct_ in cascade_states: |
| 3705 | self._delete_impl(st_, o, False) |
| 3706 | |
| 3707 | def get( |
| 3708 | self, |
no test coverage detected