(
self,
dict_: _InstanceDict,
attr: Optional[AttributeImpl],
previous: Any,
collection: bool = False,
is_userland: bool = False,
)
| 888 | ) |
| 889 | |
| 890 | def _modified_event( |
| 891 | self, |
| 892 | dict_: _InstanceDict, |
| 893 | attr: Optional[AttributeImpl], |
| 894 | previous: Any, |
| 895 | collection: bool = False, |
| 896 | is_userland: bool = False, |
| 897 | ) -> None: |
| 898 | if attr: |
| 899 | if not attr.send_modified_events: |
| 900 | return |
| 901 | if is_userland and attr.key not in dict_: |
| 902 | raise sa_exc.InvalidRequestError( |
| 903 | "Can't flag attribute '%s' modified; it's not present in " |
| 904 | "the object state" % attr.key |
| 905 | ) |
| 906 | if attr.key not in self.committed_state or is_userland: |
| 907 | if collection: |
| 908 | if TYPE_CHECKING: |
| 909 | assert is_collection_impl(attr) |
| 910 | if previous is NEVER_SET: |
| 911 | if attr.key in dict_: |
| 912 | previous = dict_[attr.key] |
| 913 | |
| 914 | if previous not in (None, NO_VALUE, NEVER_SET): |
| 915 | previous = attr.copy(previous) |
| 916 | self.committed_state[attr.key] = previous |
| 917 | |
| 918 | lkv = self._last_known_values |
| 919 | if lkv is not None and attr.key in lkv: |
| 920 | lkv[attr.key] = NO_VALUE |
| 921 | |
| 922 | # assert self._strong_obj is None or self.modified |
| 923 | |
| 924 | if (self.session_id and self._strong_obj is None) or not self.modified: |
| 925 | self.modified = True |
| 926 | instance_dict = self._instance_dict() |
| 927 | if instance_dict: |
| 928 | has_modified = bool(instance_dict._modified) |
| 929 | instance_dict._modified.add(self) |
| 930 | else: |
| 931 | has_modified = False |
| 932 | |
| 933 | # only create _strong_obj link if attached |
| 934 | # to a session |
| 935 | |
| 936 | inst = self.obj() |
| 937 | if self.session_id: |
| 938 | self._strong_obj = inst |
| 939 | |
| 940 | # if identity map already had modified objects, |
| 941 | # assume autobegin already occurred, else check |
| 942 | # for autobegin |
| 943 | if not has_modified: |
| 944 | # inline of autobegin, to ensure session transaction |
| 945 | # snapshot is established |
| 946 | try: |
| 947 | session = _sessions[self.session_id] |
no test coverage detected