Загрузить файлы в «venv/Lib/site-packages/sqlalchemy/orm»

venv/Lib/site-packages/sqlalchemy/orm/sync.py

@@ -0,0 +1,164 @@
+# orm/sync.py
+# Copyright (C) 2005-2026 the SQLAlchemy authors and contributors
+# <see AUTHORS file>
+#
+# This module is part of SQLAlchemy and is released under
+# the MIT License: https://www.opensource.org/licenses/mit-license.php
+# mypy: allow-untyped-defs, allow-untyped-calls
+
+
+"""private module containing functions used for copying data
+between instances based on join conditions.
+
+"""
+
+from __future__ import annotations
+
+from . import exc
+from . import util as orm_util
+from .base import PassiveFlag
+
+
+def populate(
+    source,
+    source_mapper,
+    dest,
+    dest_mapper,
+    synchronize_pairs,
+    uowcommit,
+    flag_cascaded_pks,
+):
+    source_dict = source.dict
+    dest_dict = dest.dict
+
+    for l, r in synchronize_pairs:
+        try:
+            # inline of source_mapper._get_state_attr_by_column
+            prop = source_mapper._columntoproperty[l]
+            value = source.manager[prop.key].impl.get(
+                source, source_dict, PassiveFlag.PASSIVE_OFF
+            )
+        except exc.UnmappedColumnError as err:
+            _raise_col_to_prop(False, source_mapper, l, dest_mapper, r, err)
+
+        try:
+            # inline of dest_mapper._set_state_attr_by_column
+            prop = dest_mapper._columntoproperty[r]
+            dest.manager[prop.key].impl.set(dest, dest_dict, value, None)
+        except exc.UnmappedColumnError as err:
+            _raise_col_to_prop(True, source_mapper, l, dest_mapper, r, err)
+
+        # technically the "r.primary_key" check isn't
+        # needed here, but we check for this condition to limit
+        # how often this logic is invoked for memory/performance
+        # reasons, since we only need this info for a primary key
+        # destination.
+        if (
+            flag_cascaded_pks
+            and l.primary_key
+            and r.primary_key
+            and r.references(l)
+        ):
+            uowcommit.attributes[("pk_cascaded", dest, r)] = True
+
+
+def bulk_populate_inherit_keys(source_dict, source_mapper, synchronize_pairs):
+    # a simplified version of populate() used by bulk insert mode
+    for l, r in synchronize_pairs:
+        try:
+            prop = source_mapper._columntoproperty[l]
+            value = source_dict[prop.key]
+        except exc.UnmappedColumnError as err:
+            _raise_col_to_prop(False, source_mapper, l, source_mapper, r, err)
+
+        try:
+            prop = source_mapper._columntoproperty[r]
+            source_dict[prop.key] = value
+        except exc.UnmappedColumnError as err:
+            _raise_col_to_prop(True, source_mapper, l, source_mapper, r, err)
+
+
+def clear(dest, dest_mapper, synchronize_pairs):
+    for l, r in synchronize_pairs:
+        if (
+            r.primary_key
+            and dest_mapper._get_state_attr_by_column(dest, dest.dict, r)
+            not in orm_util._none_set
+        ):
+            raise AssertionError(
+                f"Dependency rule on column '{l}' "
+                "tried to blank-out primary key "
+                f"column '{r}' on instance '{orm_util.state_str(dest)}'"
+            )
+        try:
+            dest_mapper._set_state_attr_by_column(dest, dest.dict, r, None)
+        except exc.UnmappedColumnError as err:
+            _raise_col_to_prop(True, None, l, dest_mapper, r, err)
+
+
+def update(source, source_mapper, dest, old_prefix, synchronize_pairs):
+    for l, r in synchronize_pairs:
+        try:
+            oldvalue = source_mapper._get_committed_attr_by_column(
+                source.obj(), l
+            )
+            value = source_mapper._get_state_attr_by_column(
+                source, source.dict, l, passive=PassiveFlag.PASSIVE_OFF
+            )
+        except exc.UnmappedColumnError as err:
+            _raise_col_to_prop(False, source_mapper, l, None, r, err)
+        dest[r.key] = value
+        dest[old_prefix + r.key] = oldvalue
+
+
+def populate_dict(source, source_mapper, dict_, synchronize_pairs):
+    for l, r in synchronize_pairs:
+        try:
+            value = source_mapper._get_state_attr_by_column(
+                source, source.dict, l, passive=PassiveFlag.PASSIVE_OFF
+            )
+        except exc.UnmappedColumnError as err:
+            _raise_col_to_prop(False, source_mapper, l, None, r, err)
+
+        dict_[r.key] = value
+
+
+def source_modified(uowcommit, source, source_mapper, synchronize_pairs):
+    """return true if the source object has changes from an old to a
+    new value on the given synchronize pairs
+
+    """
+    for l, r in synchronize_pairs:
+        try:
+            prop = source_mapper._columntoproperty[l]
+        except exc.UnmappedColumnError as err:
+            _raise_col_to_prop(False, source_mapper, l, None, r, err)
+        history = uowcommit.get_attribute_history(
+            source, prop.key, PassiveFlag.PASSIVE_NO_INITIALIZE
+        )
+        if bool(history.deleted):
+            return True
+    else:
+        return False
+
+
+def _raise_col_to_prop(
+    isdest, source_mapper, source_column, dest_mapper, dest_column, err
+):
+    if isdest:
+        raise exc.UnmappedColumnError(
+            "Can't execute sync rule for "
+            "destination column '%s'; mapper '%s' does not map "
+            "this column.  Try using an explicit `foreign_keys` "
+            "collection which does not include this column (or use "
+            "a viewonly=True relation)." % (dest_column, dest_mapper)
+        ) from err
+    else:
+        raise exc.UnmappedColumnError(
+            "Can't execute sync rule for "
+            "source column '%s'; mapper '%s' does not map this "
+            "column.  Try using an explicit `foreign_keys` "
+            "collection which does not include destination column "
+            "'%s' (or use a viewonly=True relation)."
+            % (source_column, source_mapper, dest_column)
+        ) from err

venv/Lib/site-packages/sqlalchemy/orm/unitofwork.py

@@ -0,0 +1,795 @@
+# orm/unitofwork.py
+# Copyright (C) 2005-2026 the SQLAlchemy authors and contributors
+# <see AUTHORS file>
+#
+# This module is part of SQLAlchemy and is released under
+# the MIT License: https://www.opensource.org/licenses/mit-license.php
+# mypy: ignore-errors
+
+
+"""The internals for the unit of work system.
+
+The session's flush() process passes objects to a contextual object
+here, which assembles flush tasks based on mappers and their properties,
+organizes them in order of dependency, and executes.
+
+"""
+
+from __future__ import annotations
+
+from typing import Any
+from typing import Dict
+from typing import Optional
+from typing import Set
+from typing import TYPE_CHECKING
+
+from . import attributes
+from . import exc as orm_exc
+from . import util as orm_util
+from .. import event
+from .. import util
+from ..util import topological
+
+if TYPE_CHECKING:
+    from .dependency import DependencyProcessor
+    from .interfaces import MapperProperty
+    from .mapper import Mapper
+    from .session import Session
+    from .session import SessionTransaction
+    from .state import InstanceState
+
+
+def track_cascade_events(descriptor, prop):
+    """Establish event listeners on object attributes which handle
+    cascade-on-set/append.
+
+    """
+    key = prop.key
+
+    def append(state, item, initiator, **kw):
+        # process "save_update" cascade rules for when
+        # an instance is appended to the list of another instance
+
+        if item is None:
+            return
+
+        sess = state.session
+        if sess:
+            if sess._warn_on_events:
+                sess._flush_warning("collection append")
+
+            prop = state.manager.mapper._props[key]
+            item_state = attributes.instance_state(item)
+
+            if (
+                prop._cascade.save_update
+                and (key == initiator.key)
+                and not sess._contains_state(item_state)
+            ):
+                sess._save_or_update_state(item_state)
+        return item
+
+    def remove(state, item, initiator, **kw):
+        if item is None:
+            return
+
+        sess = state.session
+
+        prop = state.manager.mapper._props[key]
+
+        if sess and sess._warn_on_events:
+            sess._flush_warning(
+                "collection remove"
+                if prop.uselist
+                else "related attribute delete"
+            )
+
+        if (
+            item is not None
+            and item is not attributes.NEVER_SET
+            and item is not attributes.PASSIVE_NO_RESULT
+            and prop._cascade.delete_orphan
+        ):
+            # expunge pending orphans
+            item_state = attributes.instance_state(item)
+
+            if prop.mapper._is_orphan(item_state):
+                if sess and item_state in sess._new:
+                    sess.expunge(item)
+                else:
+                    # the related item may or may not itself be in a
+                    # Session, however the parent for which we are catching
+                    # the event is not in a session, so memoize this on the
+                    # item
+                    item_state._orphaned_outside_of_session = True
+
+    def set_(state, newvalue, oldvalue, initiator, **kw):
+        # process "save_update" cascade rules for when an instance
+        # is attached to another instance
+        if oldvalue is newvalue:
+            return newvalue
+
+        sess = state.session
+        if sess:
+            if sess._warn_on_events:
+                sess._flush_warning("related attribute set")
+
+            prop = state.manager.mapper._props[key]
+            if newvalue is not None:
+                newvalue_state = attributes.instance_state(newvalue)
+                if (
+                    prop._cascade.save_update
+                    and (key == initiator.key)
+                    and not sess._contains_state(newvalue_state)
+                ):
+                    sess._save_or_update_state(newvalue_state)
+
+            if (
+                oldvalue is not None
+                and oldvalue is not attributes.NEVER_SET
+                and oldvalue is not attributes.PASSIVE_NO_RESULT
+                and prop._cascade.delete_orphan
+            ):
+                # possible to reach here with attributes.NEVER_SET ?
+                oldvalue_state = attributes.instance_state(oldvalue)
+
+                if oldvalue_state in sess._new and prop.mapper._is_orphan(
+                    oldvalue_state
+                ):
+                    sess.expunge(oldvalue)
+        return newvalue
+
+    event.listen(
+        descriptor, "append_wo_mutation", append, raw=True, include_key=True
+    )
+    event.listen(
+        descriptor, "append", append, raw=True, retval=True, include_key=True
+    )
+    event.listen(
+        descriptor, "remove", remove, raw=True, retval=True, include_key=True
+    )
+    event.listen(
+        descriptor, "set", set_, raw=True, retval=True, include_key=True
+    )
+
+
+class UOWTransaction:
+    session: Session
+    transaction: SessionTransaction
+    attributes: Dict[str, Any]
+    deps: util.defaultdict[Mapper[Any], Set[DependencyProcessor]]
+    mappers: util.defaultdict[Mapper[Any], Set[InstanceState[Any]]]
+
+    def __init__(self, session: Session):
+        self.session = session
+
+        # dictionary used by external actors to
+        # store arbitrary state information.
+        self.attributes = {}
+
+        # dictionary of mappers to sets of
+        # DependencyProcessors, which are also
+        # set to be part of the sorted flush actions,
+        # which have that mapper as a parent.
+        self.deps = util.defaultdict(set)
+
+        # dictionary of mappers to sets of InstanceState
+        # items pending for flush which have that mapper
+        # as a parent.
+        self.mappers = util.defaultdict(set)
+
+        # a dictionary of Preprocess objects, which gather
+        # additional states impacted by the flush
+        # and determine if a flush action is needed
+        self.presort_actions = {}
+
+        # dictionary of PostSortRec objects, each
+        # one issues work during the flush within
+        # a certain ordering.
+        self.postsort_actions = {}
+
+        # a set of 2-tuples, each containing two
+        # PostSortRec objects where the second
+        # is dependent on the first being executed
+        # first
+        self.dependencies = set()
+
+        # dictionary of InstanceState-> (isdelete, listonly)
+        # tuples, indicating if this state is to be deleted
+        # or insert/updated, or just refreshed
+        self.states = {}
+
+        # tracks InstanceStates which will be receiving
+        # a "post update" call.  Keys are mappers,
+        # values are a set of states and a set of the
+        # columns which should be included in the update.
+        self.post_update_states = util.defaultdict(lambda: (set(), set()))
+
+    @property
+    def has_work(self):
+        return bool(self.states)
+
+    def was_already_deleted(self, state):
+        """Return ``True`` if the given state is expired and was deleted
+        previously.
+        """
+        if state.expired:
+            try:
+                state._load_expired(state, attributes.PASSIVE_OFF)
+            except orm_exc.ObjectDeletedError:
+                self.session._remove_newly_deleted([state])
+                return True
+        return False
+
+    def is_deleted(self, state):
+        """Return ``True`` if the given state is marked as deleted
+        within this uowtransaction."""
+
+        return state in self.states and self.states[state][0]
+
+    def memo(self, key, callable_):
+        if key in self.attributes:
+            return self.attributes[key]
+        else:
+            self.attributes[key] = ret = callable_()
+            return ret
+
+    def remove_state_actions(self, state):
+        """Remove pending actions for a state from the uowtransaction."""
+
+        isdelete = self.states[state][0]
+
+        self.states[state] = (isdelete, True)
+
+    def get_attribute_history(
+        self, state, key, passive=attributes.PASSIVE_NO_INITIALIZE
+    ):
+        """Facade to attributes.get_state_history(), including
+        caching of results."""
+
+        hashkey = ("history", state, key)
+
+        # cache the objects, not the states; the strong reference here
+        # prevents newly loaded objects from being dereferenced during the
+        # flush process
+
+        if hashkey in self.attributes:
+            history, state_history, cached_passive = self.attributes[hashkey]
+            # if the cached lookup was "passive" and now
+            # we want non-passive, do a non-passive lookup and re-cache
+
+            if (
+                not cached_passive & attributes.SQL_OK
+                and passive & attributes.SQL_OK
+            ):
+                impl = state.manager[key].impl
+                history = impl.get_history(
+                    state,
+                    state.dict,
+                    attributes.PASSIVE_OFF
+                    | attributes.LOAD_AGAINST_COMMITTED
+                    | attributes.NO_RAISE,
+                )
+                if history and impl.uses_objects:
+                    state_history = history.as_state()
+                else:
+                    state_history = history
+                self.attributes[hashkey] = (history, state_history, passive)
+        else:
+            impl = state.manager[key].impl
+            # TODO: store the history as (state, object) tuples
+            # so we don't have to keep converting here
+            history = impl.get_history(
+                state,
+                state.dict,
+                passive
+                | attributes.LOAD_AGAINST_COMMITTED
+                | attributes.NO_RAISE,
+            )
+            if history and impl.uses_objects:
+                state_history = history.as_state()
+            else:
+                state_history = history
+            self.attributes[hashkey] = (history, state_history, passive)
+
+        return state_history
+
+    def has_dep(self, processor):
+        return (processor, True) in self.presort_actions
+
+    def register_preprocessor(self, processor, fromparent):
+        key = (processor, fromparent)
+        if key not in self.presort_actions:
+            self.presort_actions[key] = Preprocess(processor, fromparent)
+
+    def register_object(
+        self,
+        state: InstanceState[Any],
+        isdelete: bool = False,
+        listonly: bool = False,
+        cancel_delete: bool = False,
+        operation: Optional[str] = None,
+        prop: Optional[MapperProperty] = None,
+    ) -> bool:
+        if not self.session._contains_state(state):
+            # this condition is normal when objects are registered
+            # as part of a relationship cascade operation.  it should
+            # not occur for the top-level register from Session.flush().
+            if not state.deleted and operation is not None:
+                util.warn(
+                    "Object of type %s not in session, %s operation "
+                    "along '%s' will not proceed"
+                    % (orm_util.state_class_str(state), operation, prop)
+                )
+            return False
+
+        if state not in self.states:
+            mapper = state.manager.mapper
+
+            if mapper not in self.mappers:
+                self._per_mapper_flush_actions(mapper)
+
+            self.mappers[mapper].add(state)
+            self.states[state] = (isdelete, listonly)
+        else:
+            if not listonly and (isdelete or cancel_delete):
+                self.states[state] = (isdelete, False)
+        return True
+
+    def register_post_update(self, state, post_update_cols):
+        mapper = state.manager.mapper.base_mapper
+        states, cols = self.post_update_states[mapper]
+        states.add(state)
+        cols.update(post_update_cols)
+
+    def _per_mapper_flush_actions(self, mapper):
+        saves = SaveUpdateAll(self, mapper.base_mapper)
+        deletes = DeleteAll(self, mapper.base_mapper)
+        self.dependencies.add((saves, deletes))
+
+        for dep in mapper._dependency_processors:
+            dep.per_property_preprocessors(self)
+
+        for prop in mapper.relationships:
+            if prop.viewonly:
+                continue
+            dep = prop._dependency_processor
+            dep.per_property_preprocessors(self)
+
+    @util.memoized_property
+    def _mapper_for_dep(self):
+        """return a dynamic mapping of (Mapper, DependencyProcessor) to
+        True or False, indicating if the DependencyProcessor operates
+        on objects of that Mapper.
+
+        The result is stored in the dictionary persistently once
+        calculated.
+
+        """
+        return util.PopulateDict(
+            lambda tup: tup[0]._props.get(tup[1].key) is tup[1].prop
+        )
+
+    def filter_states_for_dep(self, dep, states):
+        """Filter the given list of InstanceStates to those relevant to the
+        given DependencyProcessor.
+
+        """
+        mapper_for_dep = self._mapper_for_dep
+        return [s for s in states if mapper_for_dep[(s.manager.mapper, dep)]]
+
+    def states_for_mapper_hierarchy(self, mapper, isdelete, listonly):
+        checktup = (isdelete, listonly)
+        for mapper in mapper.base_mapper.self_and_descendants:
+            for state in self.mappers[mapper]:
+                if self.states[state] == checktup:
+                    yield state
+
+    def _generate_actions(self):
+        """Generate the full, unsorted collection of PostSortRecs as
+        well as dependency pairs for this UOWTransaction.
+
+        """
+        # execute presort_actions, until all states
+        # have been processed.   a presort_action might
+        # add new states to the uow.
+        while True:
+            ret = False
+            for action in list(self.presort_actions.values()):
+                if action.execute(self):
+                    ret = True
+            if not ret:
+                break
+
+        # see if the graph of mapper dependencies has cycles.
+        self.cycles = cycles = topological.find_cycles(
+            self.dependencies, list(self.postsort_actions.values())
+        )
+
+        if cycles:
+            # if yes, break the per-mapper actions into
+            # per-state actions
+            convert = {
+                rec: set(rec.per_state_flush_actions(self)) for rec in cycles
+            }
+
+            # rewrite the existing dependencies to point to
+            # the per-state actions for those per-mapper actions
+            # that were broken up.
+            for edge in list(self.dependencies):
+                if (
+                    None in edge
+                    or edge[0].disabled
+                    or edge[1].disabled
+                    or cycles.issuperset(edge)
+                ):
+                    self.dependencies.remove(edge)
+                elif edge[0] in cycles:
+                    self.dependencies.remove(edge)
+                    for dep in convert[edge[0]]:
+                        self.dependencies.add((dep, edge[1]))
+                elif edge[1] in cycles:
+                    self.dependencies.remove(edge)
+                    for dep in convert[edge[1]]:
+                        self.dependencies.add((edge[0], dep))
+
+        return {
+            a for a in self.postsort_actions.values() if not a.disabled
+        }.difference(cycles)
+
+    def execute(self) -> None:
+        postsort_actions = self._generate_actions()
+
+        postsort_actions = sorted(
+            postsort_actions,
+            key=lambda item: item.sort_key,
+        )
+        # sort = topological.sort(self.dependencies, postsort_actions)
+        # print "--------------"
+        # print "\ndependencies:", self.dependencies
+        # print "\ncycles:", self.cycles
+        # print "\nsort:", list(sort)
+        # print "\nCOUNT OF POSTSORT ACTIONS", len(postsort_actions)
+
+        # execute
+        if self.cycles:
+            for subset in topological.sort_as_subsets(
+                self.dependencies, postsort_actions
+            ):
+                set_ = set(subset)
+                while set_:
+                    n = set_.pop()
+                    n.execute_aggregate(self, set_)
+        else:
+            for rec in topological.sort(self.dependencies, postsort_actions):
+                rec.execute(self)
+
+    def finalize_flush_changes(self) -> None:
+        """Mark processed objects as clean / deleted after a successful
+        flush().
+
+        This method is called within the flush() method after the
+        execute() method has succeeded and the transaction has been committed.
+
+        """
+        if not self.states:
+            return
+
+        states = set(self.states)
+        isdel = {
+            s for (s, (isdelete, listonly)) in self.states.items() if isdelete
+        }
+        other = states.difference(isdel)
+        if isdel:
+            self.session._remove_newly_deleted(isdel)
+        if other:
+            self.session._register_persistent(other)
+
+
+class IterateMappersMixin:
+    __slots__ = ()
+
+    def _mappers(self, uow):
+        if self.fromparent:
+            return iter(
+                m
+                for m in self.dependency_processor.parent.self_and_descendants
+                if uow._mapper_for_dep[(m, self.dependency_processor)]
+            )
+        else:
+            return self.dependency_processor.mapper.self_and_descendants
+
+
+class Preprocess(IterateMappersMixin):
+    __slots__ = (
+        "dependency_processor",
+        "fromparent",
+        "processed",
+        "setup_flush_actions",
+    )
+
+    def __init__(self, dependency_processor, fromparent):
+        self.dependency_processor = dependency_processor
+        self.fromparent = fromparent
+        self.processed = set()
+        self.setup_flush_actions = False
+
+    def execute(self, uow):
+        delete_states = set()
+        save_states = set()
+
+        for mapper in self._mappers(uow):
+            for state in uow.mappers[mapper].difference(self.processed):
+                isdelete, listonly = uow.states[state]
+                if not listonly:
+                    if isdelete:
+                        delete_states.add(state)
+                    else:
+                        save_states.add(state)
+
+        if delete_states:
+            self.dependency_processor.presort_deletes(uow, delete_states)
+            self.processed.update(delete_states)
+        if save_states:
+            self.dependency_processor.presort_saves(uow, save_states)
+            self.processed.update(save_states)
+
+        if delete_states or save_states:
+            if not self.setup_flush_actions and (
+                self.dependency_processor.prop_has_changes(
+                    uow, delete_states, True
+                )
+                or self.dependency_processor.prop_has_changes(
+                    uow, save_states, False
+                )
+            ):
+                self.dependency_processor.per_property_flush_actions(uow)
+                self.setup_flush_actions = True
+            return True
+        else:
+            return False
+
+
+class PostSortRec:
+    __slots__ = ("disabled",)
+
+    def __new__(cls, uow, *args):
+        key = (cls,) + args
+        if key in uow.postsort_actions:
+            return uow.postsort_actions[key]
+        else:
+            uow.postsort_actions[key] = ret = object.__new__(cls)
+            ret.disabled = False
+            return ret
+
+    def execute_aggregate(self, uow, recs):
+        self.execute(uow)
+
+
+class ProcessAll(IterateMappersMixin, PostSortRec):
+    __slots__ = "dependency_processor", "isdelete", "fromparent", "sort_key"
+
+    def __init__(self, uow, dependency_processor, isdelete, fromparent):
+        self.dependency_processor = dependency_processor
+        self.sort_key = (
+            "ProcessAll",
+            self.dependency_processor.sort_key,
+            isdelete,
+        )
+        self.isdelete = isdelete
+        self.fromparent = fromparent
+        uow.deps[dependency_processor.parent.base_mapper].add(
+            dependency_processor
+        )
+
+    def execute(self, uow):
+        states = self._elements(uow)
+        if self.isdelete:
+            self.dependency_processor.process_deletes(uow, states)
+        else:
+            self.dependency_processor.process_saves(uow, states)
+
+    def per_state_flush_actions(self, uow):
+        # this is handled by SaveUpdateAll and DeleteAll,
+        # since a ProcessAll should unconditionally be pulled
+        # into per-state if either the parent/child mappers
+        # are part of a cycle
+        return iter([])
+
+    def __repr__(self):
+        return "%s(%s, isdelete=%s)" % (
+            self.__class__.__name__,
+            self.dependency_processor,
+            self.isdelete,
+        )
+
+    def _elements(self, uow):
+        for mapper in self._mappers(uow):
+            for state in uow.mappers[mapper]:
+                isdelete, listonly = uow.states[state]
+                if isdelete == self.isdelete and not listonly:
+                    yield state
+
+
+class PostUpdateAll(PostSortRec):
+    __slots__ = "mapper", "isdelete", "sort_key"
+
+    def __init__(self, uow, mapper, isdelete):
+        self.mapper = mapper
+        self.isdelete = isdelete
+        self.sort_key = ("PostUpdateAll", mapper._sort_key, isdelete)
+
+    @util.preload_module("sqlalchemy.orm.persistence")
+    def execute(self, uow):
+        persistence = util.preloaded.orm_persistence
+        states, cols = uow.post_update_states[self.mapper]
+        states = [s for s in states if uow.states[s][0] == self.isdelete]
+
+        persistence.post_update(self.mapper, states, uow, cols)
+
+
+class SaveUpdateAll(PostSortRec):
+    __slots__ = ("mapper", "sort_key")
+
+    def __init__(self, uow, mapper):
+        self.mapper = mapper
+        self.sort_key = ("SaveUpdateAll", mapper._sort_key)
+        assert mapper is mapper.base_mapper
+
+    @util.preload_module("sqlalchemy.orm.persistence")
+    def execute(self, uow):
+        util.preloaded.orm_persistence.save_obj(
+            self.mapper,
+            uow.states_for_mapper_hierarchy(self.mapper, False, False),
+            uow,
+        )
+
+    def per_state_flush_actions(self, uow):
+        states = list(
+            uow.states_for_mapper_hierarchy(self.mapper, False, False)
+        )
+        base_mapper = self.mapper.base_mapper
+        delete_all = DeleteAll(uow, base_mapper)
+        for state in states:
+            # keep saves before deletes -
+            # this ensures 'row switch' operations work
+            action = SaveUpdateState(uow, state)
+            uow.dependencies.add((action, delete_all))
+            yield action
+
+        for dep in uow.deps[self.mapper]:
+            states_for_prop = uow.filter_states_for_dep(dep, states)
+            dep.per_state_flush_actions(uow, states_for_prop, False)
+
+    def __repr__(self):
+        return "%s(%s)" % (self.__class__.__name__, self.mapper)
+
+
+class DeleteAll(PostSortRec):
+    __slots__ = ("mapper", "sort_key")
+
+    def __init__(self, uow, mapper):
+        self.mapper = mapper
+        self.sort_key = ("DeleteAll", mapper._sort_key)
+        assert mapper is mapper.base_mapper
+
+    @util.preload_module("sqlalchemy.orm.persistence")
+    def execute(self, uow):
+        util.preloaded.orm_persistence.delete_obj(
+            self.mapper,
+            uow.states_for_mapper_hierarchy(self.mapper, True, False),
+            uow,
+        )
+
+    def per_state_flush_actions(self, uow):
+        states = list(
+            uow.states_for_mapper_hierarchy(self.mapper, True, False)
+        )
+        base_mapper = self.mapper.base_mapper
+        save_all = SaveUpdateAll(uow, base_mapper)
+        for state in states:
+            # keep saves before deletes -
+            # this ensures 'row switch' operations work
+            action = DeleteState(uow, state)
+            uow.dependencies.add((save_all, action))
+            yield action
+
+        for dep in uow.deps[self.mapper]:
+            states_for_prop = uow.filter_states_for_dep(dep, states)
+            dep.per_state_flush_actions(uow, states_for_prop, True)
+
+    def __repr__(self):
+        return "%s(%s)" % (self.__class__.__name__, self.mapper)
+
+
+class ProcessState(PostSortRec):
+    __slots__ = "dependency_processor", "isdelete", "state", "sort_key"
+
+    def __init__(self, uow, dependency_processor, isdelete, state):
+        self.dependency_processor = dependency_processor
+        self.sort_key = ("ProcessState", dependency_processor.sort_key)
+        self.isdelete = isdelete
+        self.state = state
+
+    def execute_aggregate(self, uow, recs):
+        cls_ = self.__class__
+        dependency_processor = self.dependency_processor
+        isdelete = self.isdelete
+        our_recs = [
+            r
+            for r in recs
+            if r.__class__ is cls_
+            and r.dependency_processor is dependency_processor
+            and r.isdelete is isdelete
+        ]
+        recs.difference_update(our_recs)
+        states = [self.state] + [r.state for r in our_recs]
+        if isdelete:
+            dependency_processor.process_deletes(uow, states)
+        else:
+            dependency_processor.process_saves(uow, states)
+
+    def __repr__(self):
+        return "%s(%s, %s, delete=%s)" % (
+            self.__class__.__name__,
+            self.dependency_processor,
+            orm_util.state_str(self.state),
+            self.isdelete,
+        )
+
+
+class SaveUpdateState(PostSortRec):
+    __slots__ = "state", "mapper", "sort_key"
+
+    def __init__(self, uow, state):
+        self.state = state
+        self.mapper = state.mapper.base_mapper
+        self.sort_key = ("ProcessState", self.mapper._sort_key)
+
+    @util.preload_module("sqlalchemy.orm.persistence")
+    def execute_aggregate(self, uow, recs):
+        persistence = util.preloaded.orm_persistence
+        cls_ = self.__class__
+        mapper = self.mapper
+        our_recs = [
+            r for r in recs if r.__class__ is cls_ and r.mapper is mapper
+        ]
+        recs.difference_update(our_recs)
+        persistence.save_obj(
+            mapper, [self.state] + [r.state for r in our_recs], uow
+        )
+
+    def __repr__(self):
+        return "%s(%s)" % (
+            self.__class__.__name__,
+            orm_util.state_str(self.state),
+        )
+
+
+class DeleteState(PostSortRec):
+    __slots__ = "state", "mapper", "sort_key"
+
+    def __init__(self, uow, state):
+        self.state = state
+        self.mapper = state.mapper.base_mapper
+        self.sort_key = ("DeleteState", self.mapper._sort_key)
+
+    @util.preload_module("sqlalchemy.orm.persistence")
+    def execute_aggregate(self, uow, recs):
+        persistence = util.preloaded.orm_persistence
+        cls_ = self.__class__
+        mapper = self.mapper
+        our_recs = [
+            r for r in recs if r.__class__ is cls_ and r.mapper is mapper
+        ]
+        recs.difference_update(our_recs)
+        states = [self.state] + [r.state for r in our_recs]
+        persistence.delete_obj(
+            mapper, [s for s in states if uow.states[s][0]], uow
+        )
+
+    def __repr__(self):
+        return "%s(%s)" % (
+            self.__class__.__name__,
+            orm_util.state_str(self.state),
+        )

venv/Lib/site-packages/sqlalchemy/orm/writeonly.py

@@ -0,0 +1,674 @@
+# orm/writeonly.py
+# Copyright (C) 2005-2026 the SQLAlchemy authors and contributors
+# <see AUTHORS file>
+#
+# This module is part of SQLAlchemy and is released under
+# the MIT License: https://www.opensource.org/licenses/mit-license.php
+
+"""Write-only collection API.
+
+This is an alternate mapped attribute style that only supports single-item
+collection mutation operations.   To read the collection, a select()
+object must be executed each time.
+
+.. versionadded:: 2.0
+
+
+"""
+
+from __future__ import annotations
+
+from typing import Any
+from typing import Collection
+from typing import Dict
+from typing import Generic
+from typing import Iterable
+from typing import Iterator
+from typing import List
+from typing import NoReturn
+from typing import Optional
+from typing import overload
+from typing import Tuple
+from typing import Type
+from typing import TYPE_CHECKING
+from typing import TypeVar
+from typing import Union
+
+from sqlalchemy.sql import bindparam
+from . import attributes
+from . import interfaces
+from . import relationships
+from . import strategies
+from .base import NEVER_SET
+from .base import object_mapper
+from .base import PassiveFlag
+from .base import RelationshipDirection
+from .. import exc
+from .. import inspect
+from .. import log
+from .. import util
+from ..sql import delete
+from ..sql import insert
+from ..sql import select
+from ..sql import update
+from ..sql.dml import Delete
+from ..sql.dml import Insert
+from ..sql.dml import Update
+from ..util.typing import Literal
+
+if TYPE_CHECKING:
+    from . import QueryableAttribute
+    from ._typing import _InstanceDict
+    from .attributes import AttributeEventToken
+    from .base import LoaderCallableStatus
+    from .collections import _AdaptedCollectionProtocol
+    from .collections import CollectionAdapter
+    from .mapper import Mapper
+    from .relationships import _RelationshipOrderByArg
+    from .state import InstanceState
+    from .util import AliasedClass
+    from ..event import _Dispatch
+    from ..sql.selectable import FromClause
+    from ..sql.selectable import Select
+
+_T = TypeVar("_T", bound=Any)
+
+
+class WriteOnlyHistory(Generic[_T]):
+    """Overrides AttributeHistory to receive append/remove events directly."""
+
+    unchanged_items: util.OrderedIdentitySet
+    added_items: util.OrderedIdentitySet
+    deleted_items: util.OrderedIdentitySet
+    _reconcile_collection: bool
+
+    def __init__(
+        self,
+        attr: WriteOnlyAttributeImpl,
+        state: InstanceState[_T],
+        passive: PassiveFlag,
+        apply_to: Optional[WriteOnlyHistory[_T]] = None,
+    ) -> None:
+        if apply_to:
+            if passive & PassiveFlag.SQL_OK:
+                raise exc.InvalidRequestError(
+                    f"Attribute {attr} can't load the existing state from the "
+                    "database for this operation; full iteration is not "
+                    "permitted.  If this is a delete operation, configure "
+                    f"passive_deletes=True on the {attr} relationship in "
+                    "order to resolve this error."
+                )
+
+            self.unchanged_items = apply_to.unchanged_items
+            self.added_items = apply_to.added_items
+            self.deleted_items = apply_to.deleted_items
+            self._reconcile_collection = apply_to._reconcile_collection
+        else:
+            self.deleted_items = util.OrderedIdentitySet()
+            self.added_items = util.OrderedIdentitySet()
+            self.unchanged_items = util.OrderedIdentitySet()
+            self._reconcile_collection = False
+
+    @property
+    def added_plus_unchanged(self) -> List[_T]:
+        return list(self.added_items.union(self.unchanged_items))
+
+    @property
+    def all_items(self) -> List[_T]:
+        return list(
+            self.added_items.union(self.unchanged_items).union(
+                self.deleted_items
+            )
+        )
+
+    def as_history(self) -> attributes.History:
+        if self._reconcile_collection:
+            added = self.added_items.difference(self.unchanged_items)
+            deleted = self.deleted_items.intersection(self.unchanged_items)
+            unchanged = self.unchanged_items.difference(deleted)
+        else:
+            added, unchanged, deleted = (
+                self.added_items,
+                self.unchanged_items,
+                self.deleted_items,
+            )
+        return attributes.History(list(added), list(unchanged), list(deleted))
+
+    def indexed(self, index: Union[int, slice]) -> Union[List[_T], _T]:
+        return list(self.added_items)[index]
+
+    def add_added(self, value: _T) -> None:
+        self.added_items.add(value)
+
+    def add_removed(self, value: _T) -> None:
+        if value in self.added_items:
+            self.added_items.remove(value)
+        else:
+            self.deleted_items.add(value)
+
+
+class WriteOnlyAttributeImpl(
+    attributes.HasCollectionAdapter, attributes.AttributeImpl
+):
+    uses_objects: bool = True
+    default_accepts_scalar_loader: bool = False
+    supports_population: bool = False
+    _supports_dynamic_iteration: bool = False
+    collection: bool = False
+    dynamic: bool = True
+    order_by: _RelationshipOrderByArg = ()
+    collection_history_cls: Type[WriteOnlyHistory[Any]] = WriteOnlyHistory
+
+    query_class: Type[WriteOnlyCollection[Any]]
+
+    def __init__(
+        self,
+        class_: Union[Type[Any], AliasedClass[Any]],
+        key: str,
+        dispatch: _Dispatch[QueryableAttribute[Any]],
+        target_mapper: Mapper[_T],
+        order_by: _RelationshipOrderByArg,
+        **kw: Any,
+    ):
+        super().__init__(class_, key, None, dispatch, **kw)
+        self.target_mapper = target_mapper
+        self.query_class = WriteOnlyCollection
+        if order_by:
+            self.order_by = tuple(order_by)
+
+    def get(
+        self,
+        state: InstanceState[Any],
+        dict_: _InstanceDict,
+        passive: PassiveFlag = PassiveFlag.PASSIVE_OFF,
+    ) -> Union[util.OrderedIdentitySet, WriteOnlyCollection[Any]]:
+        if not passive & PassiveFlag.SQL_OK:
+            return self._get_collection_history(
+                state, PassiveFlag.PASSIVE_NO_INITIALIZE
+            ).added_items
+        else:
+            return self.query_class(self, state)
+
+    @overload
+    def get_collection(
+        self,
+        state: InstanceState[Any],
+        dict_: _InstanceDict,
+        user_data: Literal[None] = ...,
+        passive: Literal[PassiveFlag.PASSIVE_OFF] = ...,
+    ) -> CollectionAdapter: ...
+
+    @overload
+    def get_collection(
+        self,
+        state: InstanceState[Any],
+        dict_: _InstanceDict,
+        user_data: _AdaptedCollectionProtocol = ...,
+        passive: PassiveFlag = ...,
+    ) -> CollectionAdapter: ...
+
+    @overload
+    def get_collection(
+        self,
+        state: InstanceState[Any],
+        dict_: _InstanceDict,
+        user_data: Optional[_AdaptedCollectionProtocol] = ...,
+        passive: PassiveFlag = ...,
+    ) -> Union[
+        Literal[LoaderCallableStatus.PASSIVE_NO_RESULT], CollectionAdapter
+    ]: ...
+
+    def get_collection(
+        self,
+        state: InstanceState[Any],
+        dict_: _InstanceDict,
+        user_data: Optional[_AdaptedCollectionProtocol] = None,
+        passive: PassiveFlag = PassiveFlag.PASSIVE_OFF,
+    ) -> Union[
+        Literal[LoaderCallableStatus.PASSIVE_NO_RESULT], CollectionAdapter
+    ]:
+        data: Collection[Any]
+        if not passive & PassiveFlag.SQL_OK:
+            data = self._get_collection_history(state, passive).added_items
+        else:
+            history = self._get_collection_history(state, passive)
+            data = history.added_plus_unchanged
+        return DynamicCollectionAdapter(data)  # type: ignore[return-value]
+
+    @util.memoized_property
+    def _append_token(self) -> attributes.AttributeEventToken:
+        return attributes.AttributeEventToken(self, attributes.OP_APPEND)
+
+    @util.memoized_property
+    def _remove_token(self) -> attributes.AttributeEventToken:
+        return attributes.AttributeEventToken(self, attributes.OP_REMOVE)
+
+    def fire_append_event(
+        self,
+        state: InstanceState[Any],
+        dict_: _InstanceDict,
+        value: Any,
+        initiator: Optional[AttributeEventToken],
+        collection_history: Optional[WriteOnlyHistory[Any]] = None,
+    ) -> None:
+        if collection_history is None:
+            collection_history = self._modified_event(state, dict_)
+
+        collection_history.add_added(value)
+
+        for fn in self.dispatch.append:
+            value = fn(state, value, initiator or self._append_token)
+
+        if self.trackparent and value is not None:
+            self.sethasparent(attributes.instance_state(value), state, True)
+
+    def fire_remove_event(
+        self,
+        state: InstanceState[Any],
+        dict_: _InstanceDict,
+        value: Any,
+        initiator: Optional[AttributeEventToken],
+        collection_history: Optional[WriteOnlyHistory[Any]] = None,
+    ) -> None:
+        if collection_history is None:
+            collection_history = self._modified_event(state, dict_)
+
+        collection_history.add_removed(value)
+
+        if self.trackparent and value is not None:
+            self.sethasparent(attributes.instance_state(value), state, False)
+
+        for fn in self.dispatch.remove:
+            fn(state, value, initiator or self._remove_token)
+
+    def _modified_event(
+        self, state: InstanceState[Any], dict_: _InstanceDict
+    ) -> WriteOnlyHistory[Any]:
+        if self.key not in state.committed_state:
+            state.committed_state[self.key] = self.collection_history_cls(
+                self, state, PassiveFlag.PASSIVE_NO_FETCH
+            )
+
+        state._modified_event(dict_, self, NEVER_SET)
+
+        # this is a hack to allow the entities.ComparableEntity fixture
+        # to work
+        dict_[self.key] = True
+        return state.committed_state[self.key]  # type: ignore[no-any-return]
+
+    def set(
+        self,
+        state: InstanceState[Any],
+        dict_: _InstanceDict,
+        value: Any,
+        initiator: Optional[AttributeEventToken] = None,
+        passive: PassiveFlag = PassiveFlag.PASSIVE_OFF,
+        check_old: Any = None,
+        pop: bool = False,
+        _adapt: bool = True,
+    ) -> None:
+        if initiator and initiator.parent_token is self.parent_token:
+            return
+
+        if pop and value is None:
+            return
+
+        iterable = value
+        new_values = list(iterable)
+        if state.has_identity:
+            if not self._supports_dynamic_iteration:
+                raise exc.InvalidRequestError(
+                    f'Collection "{self}" does not support implicit '
+                    "iteration; collection replacement operations "
+                    "can't be used"
+                )
+            old_collection = util.IdentitySet(
+                self.get(state, dict_, passive=passive)
+            )
+
+        collection_history = self._modified_event(state, dict_)
+        if not state.has_identity:
+            old_collection = collection_history.added_items
+        else:
+            old_collection = old_collection.union(
+                collection_history.added_items
+            )
+
+        constants = old_collection.intersection(new_values)
+        additions = util.IdentitySet(new_values).difference(constants)
+        removals = old_collection.difference(constants)
+
+        for member in new_values:
+            if member in additions:
+                self.fire_append_event(
+                    state,
+                    dict_,
+                    member,
+                    None,
+                    collection_history=collection_history,
+                )
+
+        for member in removals:
+            self.fire_remove_event(
+                state,
+                dict_,
+                member,
+                None,
+                collection_history=collection_history,
+            )
+
+    def delete(self, *args: Any, **kwargs: Any) -> NoReturn:
+        raise NotImplementedError()
+
+    def set_committed_value(
+        self, state: InstanceState[Any], dict_: _InstanceDict, value: Any
+    ) -> NoReturn:
+        raise NotImplementedError(
+            "Dynamic attributes don't support collection population."
+        )
+
+    def get_history(
+        self,
+        state: InstanceState[Any],
+        dict_: _InstanceDict,
+        passive: PassiveFlag = PassiveFlag.PASSIVE_NO_FETCH,
+    ) -> attributes.History:
+        c = self._get_collection_history(state, passive)
+        return c.as_history()
+
+    def get_all_pending(
+        self,
+        state: InstanceState[Any],
+        dict_: _InstanceDict,
+        passive: PassiveFlag = PassiveFlag.PASSIVE_NO_INITIALIZE,
+    ) -> List[Tuple[InstanceState[Any], Any]]:
+        c = self._get_collection_history(state, passive)
+        return [(attributes.instance_state(x), x) for x in c.all_items]
+
+    def _get_collection_history(
+        self, state: InstanceState[Any], passive: PassiveFlag
+    ) -> WriteOnlyHistory[Any]:
+        c: WriteOnlyHistory[Any]
+        if self.key in state.committed_state:
+            c = state.committed_state[self.key]
+        else:
+            c = self.collection_history_cls(
+                self, state, PassiveFlag.PASSIVE_NO_FETCH
+            )
+
+        if state.has_identity and (passive & PassiveFlag.INIT_OK):
+            return self.collection_history_cls(
+                self, state, passive, apply_to=c
+            )
+        else:
+            return c
+
+    def append(
+        self,
+        state: InstanceState[Any],
+        dict_: _InstanceDict,
+        value: Any,
+        initiator: Optional[AttributeEventToken],
+        passive: PassiveFlag = PassiveFlag.PASSIVE_NO_FETCH,
+    ) -> None:
+        if initiator is not self:  # type: ignore[comparison-overlap]
+            self.fire_append_event(state, dict_, value, initiator)
+
+    def remove(
+        self,
+        state: InstanceState[Any],
+        dict_: _InstanceDict,
+        value: Any,
+        initiator: Optional[AttributeEventToken],
+        passive: PassiveFlag = PassiveFlag.PASSIVE_NO_FETCH,
+    ) -> None:
+        if initiator is not self:  # type: ignore[comparison-overlap]
+            self.fire_remove_event(state, dict_, value, initiator)
+
+    def pop(
+        self,
+        state: InstanceState[Any],
+        dict_: _InstanceDict,
+        value: Any,
+        initiator: Optional[AttributeEventToken],
+        passive: PassiveFlag = PassiveFlag.PASSIVE_NO_FETCH,
+    ) -> None:
+        self.remove(state, dict_, value, initiator, passive=passive)
+
+
+@log.class_logger
+@relationships.RelationshipProperty.strategy_for(lazy="write_only")
+class WriteOnlyLoader(strategies.AbstractRelationshipLoader, log.Identified):
+    impl_class = WriteOnlyAttributeImpl
+
+    def init_class_attribute(self, mapper: Mapper[Any]) -> None:
+        self.is_class_level = True
+        if not self.uselist or self.parent_property.direction not in (
+            interfaces.ONETOMANY,
+            interfaces.MANYTOMANY,
+        ):
+            raise exc.InvalidRequestError(
+                "On relationship %s, 'dynamic' loaders cannot be used with "
+                "many-to-one/one-to-one relationships and/or "
+                "uselist=False." % self.parent_property
+            )
+
+        strategies._register_attribute(  # type: ignore[no-untyped-call]
+            self.parent_property,
+            mapper,
+            useobject=True,
+            impl_class=self.impl_class,
+            target_mapper=self.parent_property.mapper,
+            order_by=self.parent_property.order_by,
+            query_class=self.parent_property.query_class,
+        )
+
+
+class DynamicCollectionAdapter:
+    """simplified CollectionAdapter for internal API consistency"""
+
+    data: Collection[Any]
+
+    def __init__(self, data: Collection[Any]):
+        self.data = data
+
+    def __iter__(self) -> Iterator[Any]:
+        return iter(self.data)
+
+    def _reset_empty(self) -> None:
+        pass
+
+    def __len__(self) -> int:
+        return len(self.data)
+
+    def __bool__(self) -> bool:
+        return True
+
+
+class AbstractCollectionWriter(Generic[_T]):
+    """Virtual collection which includes append/remove methods that synchronize
+    into the attribute event system.
+
+    """
+
+    if not TYPE_CHECKING:
+        __slots__ = ()
+
+    instance: _T
+    _from_obj: Tuple[FromClause, ...]
+
+    def __init__(self, attr: WriteOnlyAttributeImpl, state: InstanceState[_T]):
+        instance = state.obj()
+        if TYPE_CHECKING:
+            assert instance
+        self.instance = instance
+        self.attr = attr
+
+        mapper = object_mapper(instance)
+        prop = mapper._props[self.attr.key]
+
+        if prop.secondary is not None:
+            # this is a hack right now.  The Query only knows how to
+            # make subsequent joins() without a given left-hand side
+            # from self._from_obj[0].  We need to ensure prop.secondary
+            # is in the FROM.  So we purposely put the mapper selectable
+            # in _from_obj[0] to ensure a user-defined join() later on
+            # doesn't fail, and secondary is then in _from_obj[1].
+
+            # note also, we are using the official ORM-annotated selectable
+            # from __clause_element__(), see #7868
+            self._from_obj = (prop.mapper.__clause_element__(), prop.secondary)
+        else:
+            self._from_obj = ()
+
+        self._where_criteria = (
+            prop._with_parent(instance, alias_secondary=False),
+        )
+
+        if self.attr.order_by:
+            self._order_by_clauses = self.attr.order_by
+        else:
+            self._order_by_clauses = ()
+
+    def _add_all_impl(self, iterator: Iterable[_T]) -> None:
+        for item in iterator:
+            self.attr.append(
+                attributes.instance_state(self.instance),
+                attributes.instance_dict(self.instance),
+                item,
+                None,
+            )
+
+    def _remove_impl(self, item: _T) -> None:
+        self.attr.remove(
+            attributes.instance_state(self.instance),
+            attributes.instance_dict(self.instance),
+            item,
+            None,
+        )
+
+
+class WriteOnlyCollection(AbstractCollectionWriter[_T]):
+    """Write-only collection which can synchronize changes into the
+    attribute event system.
+
+    The :class:`.WriteOnlyCollection` is used in a mapping by
+    using the ``"write_only"`` lazy loading strategy with
+    :func:`_orm.relationship`.     For background on this configuration,
+    see :ref:`write_only_relationship`.
+
+    .. versionadded:: 2.0
+
+    .. seealso::
+
+        :ref:`write_only_relationship`
+
+    """
+
+    __slots__ = (
+        "instance",
+        "attr",
+        "_where_criteria",
+        "_from_obj",
+        "_order_by_clauses",
+    )
+
+    def __iter__(self) -> NoReturn:
+        raise TypeError(
+            "WriteOnly collections don't support iteration in-place; "
+            "to query for collection items, use the select() method to "
+            "produce a SQL statement and execute it with session.scalars()."
+        )
+
+    def select(self) -> Select[Tuple[_T]]:
+        """Produce a :class:`_sql.Select` construct that represents the
+        rows within this instance-local :class:`_orm.WriteOnlyCollection`.
+
+        """
+        stmt = select(self.attr.target_mapper).where(*self._where_criteria)
+        if self._from_obj:
+            stmt = stmt.select_from(*self._from_obj)
+        if self._order_by_clauses:
+            stmt = stmt.order_by(*self._order_by_clauses)
+        return stmt
+
+    def insert(self) -> Insert:
+        """For one-to-many collections, produce a :class:`_dml.Insert` which
+        will insert new rows in terms of this this instance-local
+        :class:`_orm.WriteOnlyCollection`.
+
+        This construct is only supported for a :class:`_orm.Relationship`
+        that does **not** include the :paramref:`_orm.relationship.secondary`
+        parameter.  For relationships that refer to a many-to-many table,
+        use ordinary bulk insert techniques to produce new objects, then
+        use :meth:`_orm.AbstractCollectionWriter.add_all` to associate them
+        with the collection.
+
+
+        """
+
+        state = inspect(self.instance)
+        mapper = state.mapper
+        prop = mapper._props[self.attr.key]
+
+        if prop.direction is not RelationshipDirection.ONETOMANY:
+            raise exc.InvalidRequestError(
+                "Write only bulk INSERT only supported for one-to-many "
+                "collections; for many-to-many, use a separate bulk "
+                "INSERT along with add_all()."
+            )
+
+        dict_: Dict[str, Any] = {}
+
+        for l, r in prop.synchronize_pairs:
+            fn = prop._get_attr_w_warn_on_none(
+                mapper,
+                state,
+                state.dict,
+                l,
+            )
+
+            dict_[r.key] = bindparam(None, callable_=fn)
+
+        return insert(self.attr.target_mapper).values(**dict_)
+
+    def update(self) -> Update:
+        """Produce a :class:`_dml.Update` which will refer to rows in terms
+        of this instance-local :class:`_orm.WriteOnlyCollection`.
+
+        """
+        return update(self.attr.target_mapper).where(*self._where_criteria)
+
+    def delete(self) -> Delete:
+        """Produce a :class:`_dml.Delete` which will refer to rows in terms
+        of this instance-local :class:`_orm.WriteOnlyCollection`.
+
+        """
+        return delete(self.attr.target_mapper).where(*self._where_criteria)
+
+    def add_all(self, iterator: Iterable[_T]) -> None:
+        """Add an iterable of items to this :class:`_orm.WriteOnlyCollection`.
+
+        The given items will be persisted to the database in terms of
+        the parent instance's collection on the next flush.
+
+        """
+        self._add_all_impl(iterator)
+
+    def add(self, item: _T) -> None:
+        """Add an item to this :class:`_orm.WriteOnlyCollection`.
+
+        The given item will be persisted to the database in terms of
+        the parent instance's collection on the next flush.
+
+        """
+        self._add_all_impl([item])
+
+    def remove(self, item: _T) -> None:
+        """Remove an item from this :class:`_orm.WriteOnlyCollection`.
+
+        The given item will be removed from the parent instance's collection on
+        the next flush.
+
+        """
+        self._remove_impl(item)