Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# orm/writeonly.py
2# Copyright (C) 2005-2026 the SQLAlchemy authors and contributors
3# <see AUTHORS file>
4#
5# This module is part of SQLAlchemy and is released under
6# the MIT License: https://www.opensource.org/licenses/mit-license.php
8"""Write-only collection API.
10This is an alternate mapped attribute style that only supports single-item
11collection mutation operations. To read the collection, a select()
12object must be executed each time.
14.. versionadded:: 2.0
17"""
19from __future__ import annotations
21from typing import Any
22from typing import Collection
23from typing import Dict
24from typing import Generic
25from typing import Iterable
26from typing import Iterator
27from typing import List
28from typing import Literal
29from typing import NoReturn
30from typing import Optional
31from typing import overload
32from typing import Tuple
33from typing import Type
34from typing import TYPE_CHECKING
35from typing import TypeVar
36from typing import Union
38from sqlalchemy.sql import bindparam
39from . import attributes
40from . import interfaces
41from . import relationships
42from . import strategies
43from .base import ATTR_EMPTY
44from .base import DONT_SET
45from .base import NEVER_SET
46from .base import object_mapper
47from .base import PassiveFlag
48from .base import RelationshipDirection
49from .. import exc
50from .. import inspect
51from .. import log
52from .. import util
53from ..sql import delete
54from ..sql import insert
55from ..sql import select
56from ..sql import update
57from ..sql.dml import Delete
58from ..sql.dml import Insert
59from ..sql.dml import Update
61if TYPE_CHECKING:
62 from . import QueryableAttribute
63 from ._typing import _InstanceDict
64 from .attributes import AttributeEventToken
65 from .base import LoaderCallableStatus
66 from .collections import _AdaptedCollectionProtocol
67 from .collections import CollectionAdapter
68 from .mapper import Mapper
69 from .relationships import _RelationshipOrderByArg
70 from .state import InstanceState
71 from .util import AliasedClass
72 from ..event import _Dispatch
73 from ..sql.selectable import FromClause
74 from ..sql.selectable import Select
76_T = TypeVar("_T", bound=Any)
79class WriteOnlyHistory(Generic[_T]):
80 """Overrides AttributeHistory to receive append/remove events directly."""
82 unchanged_items: util.OrderedIdentitySet
83 added_items: util.OrderedIdentitySet
84 deleted_items: util.OrderedIdentitySet
85 _reconcile_collection: bool
87 def __init__(
88 self,
89 attr: _WriteOnlyAttributeImpl,
90 state: InstanceState[_T],
91 passive: PassiveFlag,
92 apply_to: Optional[WriteOnlyHistory[_T]] = None,
93 ) -> None:
94 if apply_to:
95 if passive & PassiveFlag.SQL_OK:
96 raise exc.InvalidRequestError(
97 f"Attribute {attr} can't load the existing state from the "
98 "database for this operation; full iteration is not "
99 "permitted. If this is a delete operation, configure "
100 f"passive_deletes=True on the {attr} relationship in "
101 "order to resolve this error."
102 )
104 self.unchanged_items = apply_to.unchanged_items
105 self.added_items = apply_to.added_items
106 self.deleted_items = apply_to.deleted_items
107 self._reconcile_collection = apply_to._reconcile_collection
108 else:
109 self.deleted_items = util.OrderedIdentitySet()
110 self.added_items = util.OrderedIdentitySet()
111 self.unchanged_items = util.OrderedIdentitySet()
112 self._reconcile_collection = False
114 @property
115 def added_plus_unchanged(self) -> List[_T]:
116 return list(self.added_items.union(self.unchanged_items))
118 @property
119 def all_items(self) -> List[_T]:
120 return list(
121 self.added_items.union(self.unchanged_items).union(
122 self.deleted_items
123 )
124 )
126 def as_history(self) -> attributes.History:
127 if self._reconcile_collection:
128 added = self.added_items.difference(self.unchanged_items)
129 deleted = self.deleted_items.intersection(self.unchanged_items)
130 unchanged = self.unchanged_items.difference(deleted)
131 else:
132 added, unchanged, deleted = (
133 self.added_items,
134 self.unchanged_items,
135 self.deleted_items,
136 )
137 return attributes.History(list(added), list(unchanged), list(deleted))
139 def indexed(self, index: Union[int, slice]) -> Union[List[_T], _T]:
140 return list(self.added_items)[index]
142 def add_added(self, value: _T) -> None:
143 self.added_items.add(value)
145 def add_removed(self, value: _T) -> None:
146 if value in self.added_items:
147 self.added_items.remove(value)
148 else:
149 self.deleted_items.add(value)
152class _WriteOnlyAttributeImpl(
153 attributes._HasCollectionAdapter, attributes._AttributeImpl
154):
155 uses_objects: bool = True
156 default_accepts_scalar_loader: bool = False
157 supports_population: bool = False
158 _supports_dynamic_iteration: bool = False
159 collection: bool = False
160 dynamic: bool = True
161 order_by: _RelationshipOrderByArg = ()
162 collection_history_cls: Type[WriteOnlyHistory[Any]] = WriteOnlyHistory
164 query_class: Type[WriteOnlyCollection[Any]]
166 def __init__(
167 self,
168 class_: Union[Type[Any], AliasedClass[Any]],
169 key: str,
170 dispatch: _Dispatch[QueryableAttribute[Any]],
171 target_mapper: Mapper[_T],
172 order_by: _RelationshipOrderByArg,
173 **kw: Any,
174 ):
175 super().__init__(class_, key, None, dispatch, **kw)
176 self.target_mapper = target_mapper
177 self.query_class = WriteOnlyCollection
178 if order_by:
179 self.order_by = tuple(order_by)
181 def get(
182 self,
183 state: InstanceState[Any],
184 dict_: _InstanceDict,
185 passive: PassiveFlag = PassiveFlag.PASSIVE_OFF,
186 ) -> Union[util.OrderedIdentitySet, WriteOnlyCollection[Any]]:
187 if not passive & PassiveFlag.SQL_OK:
188 return self._get_collection_history(
189 state, PassiveFlag.PASSIVE_NO_INITIALIZE
190 ).added_items
191 else:
192 return self.query_class(self, state)
194 @overload
195 def get_collection(
196 self,
197 state: InstanceState[Any],
198 dict_: _InstanceDict,
199 user_data: Literal[None] = ...,
200 passive: Literal[PassiveFlag.PASSIVE_OFF] = ...,
201 ) -> CollectionAdapter: ...
203 @overload
204 def get_collection(
205 self,
206 state: InstanceState[Any],
207 dict_: _InstanceDict,
208 user_data: _AdaptedCollectionProtocol = ...,
209 passive: PassiveFlag = ...,
210 ) -> CollectionAdapter: ...
212 @overload
213 def get_collection(
214 self,
215 state: InstanceState[Any],
216 dict_: _InstanceDict,
217 user_data: Optional[_AdaptedCollectionProtocol] = ...,
218 passive: PassiveFlag = ...,
219 ) -> Union[
220 Literal[LoaderCallableStatus.PASSIVE_NO_RESULT], CollectionAdapter
221 ]: ...
223 def get_collection(
224 self,
225 state: InstanceState[Any],
226 dict_: _InstanceDict,
227 user_data: Optional[_AdaptedCollectionProtocol] = None,
228 passive: PassiveFlag = PassiveFlag.PASSIVE_OFF,
229 ) -> Union[
230 Literal[LoaderCallableStatus.PASSIVE_NO_RESULT], CollectionAdapter
231 ]:
232 data: Collection[Any]
233 if not passive & PassiveFlag.SQL_OK:
234 data = self._get_collection_history(state, passive).added_items
235 else:
236 history = self._get_collection_history(state, passive)
237 data = history.added_plus_unchanged
238 return _DynamicCollectionAdapter(data) # type: ignore[return-value]
240 @util.memoized_property
241 def _append_token(self) -> attributes.AttributeEventToken:
242 return attributes.AttributeEventToken(self, attributes.OP_APPEND)
244 @util.memoized_property
245 def _remove_token(self) -> attributes.AttributeEventToken:
246 return attributes.AttributeEventToken(self, attributes.OP_REMOVE)
248 def fire_append_event(
249 self,
250 state: InstanceState[Any],
251 dict_: _InstanceDict,
252 value: Any,
253 initiator: Optional[AttributeEventToken],
254 collection_history: Optional[WriteOnlyHistory[Any]] = None,
255 ) -> None:
256 if collection_history is None:
257 collection_history = self._modified_event(state, dict_)
259 collection_history.add_added(value)
261 for fn in self.dispatch.append:
262 value = fn(state, value, initiator or self._append_token)
264 if self.trackparent and value is not None:
265 self.sethasparent(attributes.instance_state(value), state, True)
267 def fire_remove_event(
268 self,
269 state: InstanceState[Any],
270 dict_: _InstanceDict,
271 value: Any,
272 initiator: Optional[AttributeEventToken],
273 collection_history: Optional[WriteOnlyHistory[Any]] = None,
274 ) -> None:
275 if collection_history is None:
276 collection_history = self._modified_event(state, dict_)
278 collection_history.add_removed(value)
280 if self.trackparent and value is not None:
281 self.sethasparent(attributes.instance_state(value), state, False)
283 for fn in self.dispatch.remove:
284 fn(state, value, initiator or self._remove_token)
286 def _modified_event(
287 self, state: InstanceState[Any], dict_: _InstanceDict
288 ) -> WriteOnlyHistory[Any]:
289 if self.key not in state.committed_state:
290 state.committed_state[self.key] = self.collection_history_cls(
291 self, state, PassiveFlag.PASSIVE_NO_FETCH
292 )
294 state._modified_event(dict_, self, NEVER_SET)
296 # this is a hack to allow the entities.ComparableEntity fixture
297 # to work
298 dict_[self.key] = True
299 return state.committed_state[self.key] # type: ignore[no-any-return]
301 def set(
302 self,
303 state: InstanceState[Any],
304 dict_: _InstanceDict,
305 value: Any,
306 initiator: Optional[AttributeEventToken] = None,
307 passive: PassiveFlag = PassiveFlag.PASSIVE_OFF,
308 check_old: Any = None,
309 pop: bool = False,
310 _adapt: bool = True,
311 ) -> None:
312 if initiator and initiator.parent_token is self.parent_token:
313 return
315 if pop and value is None:
316 return
318 if value is DONT_SET:
319 # dataclasses default_factory for a write only collection
320 # sends DONT_SET; there's no collection to initialize so
321 # this is a no-op
322 return
324 iterable = value
325 new_values = list(iterable)
326 if state.has_identity:
327 if not self._supports_dynamic_iteration:
328 raise exc.InvalidRequestError(
329 f'Collection "{self}" does not support implicit '
330 "iteration; collection replacement operations "
331 "can't be used"
332 )
333 old_collection = util.IdentitySet(
334 self.get(state, dict_, passive=passive)
335 )
337 collection_history = self._modified_event(state, dict_)
338 if not state.has_identity:
339 old_collection = collection_history.added_items
340 else:
341 old_collection = old_collection.union(
342 collection_history.added_items
343 )
345 constants = old_collection.intersection(new_values)
346 additions = util.IdentitySet(new_values).difference(constants)
347 removals = old_collection.difference(constants)
349 for member in new_values:
350 if member in additions:
351 self.fire_append_event(
352 state,
353 dict_,
354 member,
355 None,
356 collection_history=collection_history,
357 )
359 for member in removals:
360 self.fire_remove_event(
361 state,
362 dict_,
363 member,
364 None,
365 collection_history=collection_history,
366 )
368 def delete(self, *args: Any, **kwargs: Any) -> NoReturn:
369 raise NotImplementedError()
371 def set_committed_value(
372 self, state: InstanceState[Any], dict_: _InstanceDict, value: Any
373 ) -> NoReturn:
374 raise NotImplementedError(
375 "Dynamic attributes don't support collection population."
376 )
378 def get_history(
379 self,
380 state: InstanceState[Any],
381 dict_: _InstanceDict,
382 passive: PassiveFlag = PassiveFlag.PASSIVE_NO_FETCH,
383 ) -> attributes.History:
384 c = self._get_collection_history(state, passive)
385 return c.as_history()
387 def get_all_pending(
388 self,
389 state: InstanceState[Any],
390 dict_: _InstanceDict,
391 passive: PassiveFlag = PassiveFlag.PASSIVE_NO_INITIALIZE,
392 ) -> List[Tuple[InstanceState[Any], Any]]:
393 c = self._get_collection_history(state, passive)
394 return [(attributes.instance_state(x), x) for x in c.all_items]
396 def _default_value(
397 self, state: InstanceState[Any], dict_: _InstanceDict
398 ) -> Any:
399 value = None
400 for fn in self.dispatch.init_scalar:
401 ret = fn(state, value, dict_)
402 if ret is not ATTR_EMPTY:
403 value = ret
405 return value
407 def _get_collection_history(
408 self, state: InstanceState[Any], passive: PassiveFlag
409 ) -> WriteOnlyHistory[Any]:
410 c: WriteOnlyHistory[Any]
411 if self.key in state.committed_state:
412 c = state.committed_state[self.key]
413 else:
414 c = self.collection_history_cls(
415 self, state, PassiveFlag.PASSIVE_NO_FETCH
416 )
418 if state.has_identity and (passive & PassiveFlag.INIT_OK):
419 return self.collection_history_cls(
420 self, state, passive, apply_to=c
421 )
422 else:
423 return c
425 def append(
426 self,
427 state: InstanceState[Any],
428 dict_: _InstanceDict,
429 value: Any,
430 initiator: Optional[AttributeEventToken],
431 passive: PassiveFlag = PassiveFlag.PASSIVE_NO_FETCH,
432 ) -> None:
433 if initiator is not self: # type: ignore[comparison-overlap]
434 self.fire_append_event(state, dict_, value, initiator)
436 def remove(
437 self,
438 state: InstanceState[Any],
439 dict_: _InstanceDict,
440 value: Any,
441 initiator: Optional[AttributeEventToken],
442 passive: PassiveFlag = PassiveFlag.PASSIVE_NO_FETCH,
443 ) -> None:
444 if initiator is not self: # type: ignore[comparison-overlap]
445 self.fire_remove_event(state, dict_, value, initiator)
447 def pop(
448 self,
449 state: InstanceState[Any],
450 dict_: _InstanceDict,
451 value: Any,
452 initiator: Optional[AttributeEventToken],
453 passive: PassiveFlag = PassiveFlag.PASSIVE_NO_FETCH,
454 ) -> None:
455 self.remove(state, dict_, value, initiator, passive=passive)
458@log.class_logger
459@relationships.RelationshipProperty.strategy_for(lazy="write_only")
460class _WriteOnlyLoader(strategies._AbstractRelationshipLoader, log.Identified):
461 impl_class = _WriteOnlyAttributeImpl
463 def init_class_attribute(self, mapper: Mapper[Any]) -> None:
464 self.is_class_level = True
465 if not self.uselist or self.parent_property.direction not in (
466 interfaces.ONETOMANY,
467 interfaces.MANYTOMANY,
468 ):
469 raise exc.InvalidRequestError(
470 "On relationship %s, 'dynamic' loaders cannot be used with "
471 "many-to-one/one-to-one relationships and/or "
472 "uselist=False." % self.parent_property
473 )
475 strategies._register_attribute( # type: ignore[no-untyped-call]
476 self.parent_property,
477 mapper,
478 useobject=True,
479 impl_class=self.impl_class,
480 target_mapper=self.parent_property.mapper,
481 order_by=self.parent_property.order_by,
482 query_class=self.parent_property.query_class,
483 )
486class _DynamicCollectionAdapter:
487 """simplified CollectionAdapter for internal API consistency"""
489 data: Collection[Any]
491 def __init__(self, data: Collection[Any]):
492 self.data = data
494 def __iter__(self) -> Iterator[Any]:
495 return iter(self.data)
497 def _reset_empty(self) -> None:
498 pass
500 def __len__(self) -> int:
501 return len(self.data)
503 def __bool__(self) -> bool:
504 return True
507class _AbstractCollectionWriter(Generic[_T]):
508 """Virtual collection which includes append/remove methods that synchronize
509 into the attribute event system.
511 """
513 if not TYPE_CHECKING:
514 __slots__ = ()
516 instance: _T
517 _from_obj: Tuple[FromClause, ...]
519 def __init__(
520 self, attr: _WriteOnlyAttributeImpl, state: InstanceState[_T]
521 ):
522 instance = state.obj()
523 if TYPE_CHECKING:
524 assert instance
525 self.instance = instance
526 self.attr = attr
528 mapper = object_mapper(instance)
529 prop = mapper._props[self.attr.key]
531 if prop.secondary is not None:
532 # this is a hack right now. The Query only knows how to
533 # make subsequent joins() without a given left-hand side
534 # from self._from_obj[0]. We need to ensure prop.secondary
535 # is in the FROM. So we purposely put the mapper selectable
536 # in _from_obj[0] to ensure a user-defined join() later on
537 # doesn't fail, and secondary is then in _from_obj[1].
539 # note also, we are using the official ORM-annotated selectable
540 # from __clause_element__(), see #7868
542 # _no_filter_by annotation is to prevent this table from being
543 # considered by filter_by() as part of #8601
544 self._from_obj = (
545 prop.mapper.__clause_element__(),
546 prop.secondary._annotate({"_no_filter_by": True}),
547 )
548 else:
549 self._from_obj = ()
551 self._where_criteria = (
552 prop._with_parent(instance, alias_secondary=False),
553 )
555 if self.attr.order_by:
556 self._order_by_clauses = self.attr.order_by
557 else:
558 self._order_by_clauses = ()
560 def _add_all_impl(self, iterator: Iterable[_T]) -> None:
561 for item in iterator:
562 self.attr.append(
563 attributes.instance_state(self.instance),
564 attributes.instance_dict(self.instance),
565 item,
566 None,
567 )
569 def _remove_impl(self, item: _T) -> None:
570 self.attr.remove(
571 attributes.instance_state(self.instance),
572 attributes.instance_dict(self.instance),
573 item,
574 None,
575 )
578class WriteOnlyCollection(_AbstractCollectionWriter[_T]):
579 """Write-only collection which can synchronize changes into the
580 attribute event system.
582 The :class:`.WriteOnlyCollection` is used in a mapping by
583 using the ``"write_only"`` lazy loading strategy with
584 :func:`_orm.relationship`. For background on this configuration,
585 see :ref:`write_only_relationship`.
587 .. versionadded:: 2.0
589 .. seealso::
591 :ref:`write_only_relationship`
593 """
595 __slots__ = (
596 "instance",
597 "attr",
598 "_where_criteria",
599 "_from_obj",
600 "_order_by_clauses",
601 )
603 def __iter__(self) -> NoReturn:
604 raise TypeError(
605 "WriteOnly collections don't support iteration in-place; "
606 "to query for collection items, use the select() method to "
607 "produce a SQL statement and execute it with session.scalars()."
608 )
610 def select(self) -> Select[_T]:
611 """Produce a :class:`_sql.Select` construct that represents the
612 rows within this instance-local :class:`_orm.WriteOnlyCollection`.
614 """
615 stmt = select(self.attr.target_mapper).where(*self._where_criteria)
616 if self._from_obj:
617 stmt = stmt.select_from(*self._from_obj)
618 if self._order_by_clauses:
619 stmt = stmt.order_by(*self._order_by_clauses)
620 return stmt
622 def insert(self) -> Insert:
623 """For one-to-many collections, produce a :class:`_dml.Insert` which
624 will insert new rows in terms of this this instance-local
625 :class:`_orm.WriteOnlyCollection`.
627 This construct is only supported for a :class:`_orm.Relationship`
628 that does **not** include the :paramref:`_orm.relationship.secondary`
629 parameter. For relationships that refer to a many-to-many table,
630 use ordinary bulk insert techniques to produce new objects, then
631 use :meth:`_orm.AbstractCollectionWriter.add_all` to associate them
632 with the collection.
635 """
637 state = inspect(self.instance)
638 mapper = state.mapper
639 prop = mapper._props[self.attr.key]
641 if prop.direction is not RelationshipDirection.ONETOMANY:
642 raise exc.InvalidRequestError(
643 "Write only bulk INSERT only supported for one-to-many "
644 "collections; for many-to-many, use a separate bulk "
645 "INSERT along with add_all()."
646 )
648 dict_: Dict[str, Any] = {}
650 for l, r in prop.synchronize_pairs:
651 fn = prop._get_attr_w_warn_on_none(
652 mapper,
653 state,
654 state.dict,
655 l,
656 )
658 dict_[r.key] = bindparam(None, callable_=fn)
660 return insert(self.attr.target_mapper).values(**dict_)
662 def update(self) -> Update:
663 """Produce a :class:`_dml.Update` which will refer to rows in terms
664 of this instance-local :class:`_orm.WriteOnlyCollection`.
666 """
667 return update(self.attr.target_mapper).where(*self._where_criteria)
669 def delete(self) -> Delete:
670 """Produce a :class:`_dml.Delete` which will refer to rows in terms
671 of this instance-local :class:`_orm.WriteOnlyCollection`.
673 """
674 return delete(self.attr.target_mapper).where(*self._where_criteria)
676 def add_all(self, iterator: Iterable[_T]) -> None:
677 """Add an iterable of items to this :class:`_orm.WriteOnlyCollection`.
679 The given items will be persisted to the database in terms of
680 the parent instance's collection on the next flush.
682 """
683 self._add_all_impl(iterator)
685 def add(self, item: _T) -> None:
686 """Add an item to this :class:`_orm.WriteOnlyCollection`.
688 The given item will be persisted to the database in terms of
689 the parent instance's collection on the next flush.
691 """
692 self._add_all_impl([item])
694 def remove(self, item: _T) -> None:
695 """Remove an item from this :class:`_orm.WriteOnlyCollection`.
697 The given item will be removed from the parent instance's collection on
698 the next flush.
700 """
701 self._remove_impl(item)