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-2024 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 NoReturn
29from typing import Optional
30from typing import overload
31from typing import Tuple
32from typing import Type
33from typing import TYPE_CHECKING
34from typing import TypeVar
35from typing import Union
37from sqlalchemy.sql import bindparam
38from . import attributes
39from . import interfaces
40from . import relationships
41from . import strategies
42from .base import NEVER_SET
43from .base import object_mapper
44from .base import PassiveFlag
45from .base import RelationshipDirection
46from .. import exc
47from .. import inspect
48from .. import log
49from .. import util
50from ..sql import delete
51from ..sql import insert
52from ..sql import select
53from ..sql import update
54from ..sql.dml import Delete
55from ..sql.dml import Insert
56from ..sql.dml import Update
57from ..util.typing import Literal
59if TYPE_CHECKING:
60 from . import QueryableAttribute
61 from ._typing import _InstanceDict
62 from .attributes import AttributeEventToken
63 from .base import LoaderCallableStatus
64 from .collections import _AdaptedCollectionProtocol
65 from .collections import CollectionAdapter
66 from .mapper import Mapper
67 from .relationships import _RelationshipOrderByArg
68 from .state import InstanceState
69 from .util import AliasedClass
70 from ..event import _Dispatch
71 from ..sql.selectable import FromClause
72 from ..sql.selectable import Select
74_T = TypeVar("_T", bound=Any)
77class WriteOnlyHistory(Generic[_T]):
78 """Overrides AttributeHistory to receive append/remove events directly."""
80 unchanged_items: util.OrderedIdentitySet
81 added_items: util.OrderedIdentitySet
82 deleted_items: util.OrderedIdentitySet
83 _reconcile_collection: bool
85 def __init__(
86 self,
87 attr: WriteOnlyAttributeImpl,
88 state: InstanceState[_T],
89 passive: PassiveFlag,
90 apply_to: Optional[WriteOnlyHistory[_T]] = None,
91 ) -> None:
92 if apply_to:
93 if passive & PassiveFlag.SQL_OK:
94 raise exc.InvalidRequestError(
95 f"Attribute {attr} can't load the existing state from the "
96 "database for this operation; full iteration is not "
97 "permitted. If this is a delete operation, configure "
98 f"passive_deletes=True on the {attr} relationship in "
99 "order to resolve this error."
100 )
102 self.unchanged_items = apply_to.unchanged_items
103 self.added_items = apply_to.added_items
104 self.deleted_items = apply_to.deleted_items
105 self._reconcile_collection = apply_to._reconcile_collection
106 else:
107 self.deleted_items = util.OrderedIdentitySet()
108 self.added_items = util.OrderedIdentitySet()
109 self.unchanged_items = util.OrderedIdentitySet()
110 self._reconcile_collection = False
112 @property
113 def added_plus_unchanged(self) -> List[_T]:
114 return list(self.added_items.union(self.unchanged_items))
116 @property
117 def all_items(self) -> List[_T]:
118 return list(
119 self.added_items.union(self.unchanged_items).union(
120 self.deleted_items
121 )
122 )
124 def as_history(self) -> attributes.History:
125 if self._reconcile_collection:
126 added = self.added_items.difference(self.unchanged_items)
127 deleted = self.deleted_items.intersection(self.unchanged_items)
128 unchanged = self.unchanged_items.difference(deleted)
129 else:
130 added, unchanged, deleted = (
131 self.added_items,
132 self.unchanged_items,
133 self.deleted_items,
134 )
135 return attributes.History(list(added), list(unchanged), list(deleted))
137 def indexed(self, index: Union[int, slice]) -> Union[List[_T], _T]:
138 return list(self.added_items)[index]
140 def add_added(self, value: _T) -> None:
141 self.added_items.add(value)
143 def add_removed(self, value: _T) -> None:
144 if value in self.added_items:
145 self.added_items.remove(value)
146 else:
147 self.deleted_items.add(value)
150class WriteOnlyAttributeImpl(
151 attributes.HasCollectionAdapter, attributes.AttributeImpl
152):
153 uses_objects: bool = True
154 default_accepts_scalar_loader: bool = False
155 supports_population: bool = False
156 _supports_dynamic_iteration: bool = False
157 collection: bool = False
158 dynamic: bool = True
159 order_by: _RelationshipOrderByArg = ()
160 collection_history_cls: Type[WriteOnlyHistory[Any]] = WriteOnlyHistory
162 query_class: Type[WriteOnlyCollection[Any]]
164 def __init__(
165 self,
166 class_: Union[Type[Any], AliasedClass[Any]],
167 key: str,
168 dispatch: _Dispatch[QueryableAttribute[Any]],
169 target_mapper: Mapper[_T],
170 order_by: _RelationshipOrderByArg,
171 **kw: Any,
172 ):
173 super().__init__(class_, key, None, dispatch, **kw)
174 self.target_mapper = target_mapper
175 self.query_class = WriteOnlyCollection
176 if order_by:
177 self.order_by = tuple(order_by)
179 def get(
180 self,
181 state: InstanceState[Any],
182 dict_: _InstanceDict,
183 passive: PassiveFlag = PassiveFlag.PASSIVE_OFF,
184 ) -> Union[util.OrderedIdentitySet, WriteOnlyCollection[Any]]:
185 if not passive & PassiveFlag.SQL_OK:
186 return self._get_collection_history(
187 state, PassiveFlag.PASSIVE_NO_INITIALIZE
188 ).added_items
189 else:
190 return self.query_class(self, state)
192 @overload
193 def get_collection(
194 self,
195 state: InstanceState[Any],
196 dict_: _InstanceDict,
197 user_data: Literal[None] = ...,
198 passive: Literal[PassiveFlag.PASSIVE_OFF] = ...,
199 ) -> CollectionAdapter: ...
201 @overload
202 def get_collection(
203 self,
204 state: InstanceState[Any],
205 dict_: _InstanceDict,
206 user_data: _AdaptedCollectionProtocol = ...,
207 passive: PassiveFlag = ...,
208 ) -> CollectionAdapter: ...
210 @overload
211 def get_collection(
212 self,
213 state: InstanceState[Any],
214 dict_: _InstanceDict,
215 user_data: Optional[_AdaptedCollectionProtocol] = ...,
216 passive: PassiveFlag = ...,
217 ) -> Union[
218 Literal[LoaderCallableStatus.PASSIVE_NO_RESULT], CollectionAdapter
219 ]: ...
221 def get_collection(
222 self,
223 state: InstanceState[Any],
224 dict_: _InstanceDict,
225 user_data: Optional[_AdaptedCollectionProtocol] = None,
226 passive: PassiveFlag = PassiveFlag.PASSIVE_OFF,
227 ) -> Union[
228 Literal[LoaderCallableStatus.PASSIVE_NO_RESULT], CollectionAdapter
229 ]:
230 data: Collection[Any]
231 if not passive & PassiveFlag.SQL_OK:
232 data = self._get_collection_history(state, passive).added_items
233 else:
234 history = self._get_collection_history(state, passive)
235 data = history.added_plus_unchanged
236 return DynamicCollectionAdapter(data) # type: ignore[return-value]
238 @util.memoized_property
239 def _append_token( # type:ignore[override]
240 self,
241 ) -> attributes.AttributeEventToken:
242 return attributes.AttributeEventToken(self, attributes.OP_APPEND)
244 @util.memoized_property
245 def _remove_token( # type:ignore[override]
246 self,
247 ) -> attributes.AttributeEventToken:
248 return attributes.AttributeEventToken(self, attributes.OP_REMOVE)
250 def fire_append_event(
251 self,
252 state: InstanceState[Any],
253 dict_: _InstanceDict,
254 value: Any,
255 initiator: Optional[AttributeEventToken],
256 collection_history: Optional[WriteOnlyHistory[Any]] = None,
257 ) -> None:
258 if collection_history is None:
259 collection_history = self._modified_event(state, dict_)
261 collection_history.add_added(value)
263 for fn in self.dispatch.append:
264 value = fn(state, value, initiator or self._append_token)
266 if self.trackparent and value is not None:
267 self.sethasparent(attributes.instance_state(value), state, True)
269 def fire_remove_event(
270 self,
271 state: InstanceState[Any],
272 dict_: _InstanceDict,
273 value: Any,
274 initiator: Optional[AttributeEventToken],
275 collection_history: Optional[WriteOnlyHistory[Any]] = None,
276 ) -> None:
277 if collection_history is None:
278 collection_history = self._modified_event(state, dict_)
280 collection_history.add_removed(value)
282 if self.trackparent and value is not None:
283 self.sethasparent(attributes.instance_state(value), state, False)
285 for fn in self.dispatch.remove:
286 fn(state, value, initiator or self._remove_token)
288 def _modified_event(
289 self, state: InstanceState[Any], dict_: _InstanceDict
290 ) -> WriteOnlyHistory[Any]:
291 if self.key not in state.committed_state:
292 state.committed_state[self.key] = self.collection_history_cls(
293 self, state, PassiveFlag.PASSIVE_NO_FETCH
294 )
296 state._modified_event(dict_, self, NEVER_SET)
298 # this is a hack to allow the entities.ComparableEntity fixture
299 # to work
300 dict_[self.key] = True
301 return state.committed_state[self.key] # type: ignore[no-any-return]
303 def set(
304 self,
305 state: InstanceState[Any],
306 dict_: _InstanceDict,
307 value: Any,
308 initiator: Optional[AttributeEventToken] = None,
309 passive: PassiveFlag = PassiveFlag.PASSIVE_OFF,
310 check_old: Any = None,
311 pop: bool = False,
312 _adapt: bool = True,
313 ) -> None:
314 if initiator and initiator.parent_token is self.parent_token:
315 return
317 if pop and value is None:
318 return
320 iterable = value
321 new_values = list(iterable)
322 if state.has_identity:
323 if not self._supports_dynamic_iteration:
324 raise exc.InvalidRequestError(
325 f'Collection "{self}" does not support implicit '
326 "iteration; collection replacement operations "
327 "can't be used"
328 )
329 old_collection = util.IdentitySet(
330 self.get(state, dict_, passive=passive)
331 )
333 collection_history = self._modified_event(state, dict_)
334 if not state.has_identity:
335 old_collection = collection_history.added_items
336 else:
337 old_collection = old_collection.union(
338 collection_history.added_items
339 )
341 constants = old_collection.intersection(new_values)
342 additions = util.IdentitySet(new_values).difference(constants)
343 removals = old_collection.difference(constants)
345 for member in new_values:
346 if member in additions:
347 self.fire_append_event(
348 state,
349 dict_,
350 member,
351 None,
352 collection_history=collection_history,
353 )
355 for member in removals:
356 self.fire_remove_event(
357 state,
358 dict_,
359 member,
360 None,
361 collection_history=collection_history,
362 )
364 def delete(self, *args: Any, **kwargs: Any) -> NoReturn:
365 raise NotImplementedError()
367 def set_committed_value(
368 self, state: InstanceState[Any], dict_: _InstanceDict, value: Any
369 ) -> NoReturn:
370 raise NotImplementedError(
371 "Dynamic attributes don't support collection population."
372 )
374 def get_history(
375 self,
376 state: InstanceState[Any],
377 dict_: _InstanceDict,
378 passive: PassiveFlag = PassiveFlag.PASSIVE_NO_FETCH,
379 ) -> attributes.History:
380 c = self._get_collection_history(state, passive)
381 return c.as_history()
383 def get_all_pending(
384 self,
385 state: InstanceState[Any],
386 dict_: _InstanceDict,
387 passive: PassiveFlag = PassiveFlag.PASSIVE_NO_INITIALIZE,
388 ) -> List[Tuple[InstanceState[Any], Any]]:
389 c = self._get_collection_history(state, passive)
390 return [(attributes.instance_state(x), x) for x in c.all_items]
392 def _get_collection_history(
393 self, state: InstanceState[Any], passive: PassiveFlag
394 ) -> WriteOnlyHistory[Any]:
395 c: WriteOnlyHistory[Any]
396 if self.key in state.committed_state:
397 c = state.committed_state[self.key]
398 else:
399 c = self.collection_history_cls(
400 self, state, PassiveFlag.PASSIVE_NO_FETCH
401 )
403 if state.has_identity and (passive & PassiveFlag.INIT_OK):
404 return self.collection_history_cls(
405 self, state, passive, apply_to=c
406 )
407 else:
408 return c
410 def append(
411 self,
412 state: InstanceState[Any],
413 dict_: _InstanceDict,
414 value: Any,
415 initiator: Optional[AttributeEventToken],
416 passive: PassiveFlag = PassiveFlag.PASSIVE_NO_FETCH,
417 ) -> None:
418 if initiator is not self:
419 self.fire_append_event(state, dict_, value, initiator)
421 def remove(
422 self,
423 state: InstanceState[Any],
424 dict_: _InstanceDict,
425 value: Any,
426 initiator: Optional[AttributeEventToken],
427 passive: PassiveFlag = PassiveFlag.PASSIVE_NO_FETCH,
428 ) -> None:
429 if initiator is not self:
430 self.fire_remove_event(state, dict_, value, initiator)
432 def pop(
433 self,
434 state: InstanceState[Any],
435 dict_: _InstanceDict,
436 value: Any,
437 initiator: Optional[AttributeEventToken],
438 passive: PassiveFlag = PassiveFlag.PASSIVE_NO_FETCH,
439 ) -> None:
440 self.remove(state, dict_, value, initiator, passive=passive)
443@log.class_logger
444@relationships.RelationshipProperty.strategy_for(lazy="write_only")
445class WriteOnlyLoader(strategies.AbstractRelationshipLoader, log.Identified):
446 impl_class = WriteOnlyAttributeImpl
448 def init_class_attribute(self, mapper: Mapper[Any]) -> None:
449 self.is_class_level = True
450 if not self.uselist or self.parent_property.direction not in (
451 interfaces.ONETOMANY,
452 interfaces.MANYTOMANY,
453 ):
454 raise exc.InvalidRequestError(
455 "On relationship %s, 'dynamic' loaders cannot be used with "
456 "many-to-one/one-to-one relationships and/or "
457 "uselist=False." % self.parent_property
458 )
460 strategies._register_attribute( # type: ignore[no-untyped-call]
461 self.parent_property,
462 mapper,
463 useobject=True,
464 impl_class=self.impl_class,
465 target_mapper=self.parent_property.mapper,
466 order_by=self.parent_property.order_by,
467 query_class=self.parent_property.query_class,
468 )
471class DynamicCollectionAdapter:
472 """simplified CollectionAdapter for internal API consistency"""
474 data: Collection[Any]
476 def __init__(self, data: Collection[Any]):
477 self.data = data
479 def __iter__(self) -> Iterator[Any]:
480 return iter(self.data)
482 def _reset_empty(self) -> None:
483 pass
485 def __len__(self) -> int:
486 return len(self.data)
488 def __bool__(self) -> bool:
489 return True
492class AbstractCollectionWriter(Generic[_T]):
493 """Virtual collection which includes append/remove methods that synchronize
494 into the attribute event system.
496 """
498 if not TYPE_CHECKING:
499 __slots__ = ()
501 instance: _T
502 _from_obj: Tuple[FromClause, ...]
504 def __init__(self, attr: WriteOnlyAttributeImpl, state: InstanceState[_T]):
505 instance = state.obj()
506 if TYPE_CHECKING:
507 assert instance
508 self.instance = instance
509 self.attr = attr
511 mapper = object_mapper(instance)
512 prop = mapper._props[self.attr.key]
514 if prop.secondary is not None:
515 # this is a hack right now. The Query only knows how to
516 # make subsequent joins() without a given left-hand side
517 # from self._from_obj[0]. We need to ensure prop.secondary
518 # is in the FROM. So we purposely put the mapper selectable
519 # in _from_obj[0] to ensure a user-defined join() later on
520 # doesn't fail, and secondary is then in _from_obj[1].
522 # note also, we are using the official ORM-annotated selectable
523 # from __clause_element__(), see #7868
524 self._from_obj = (prop.mapper.__clause_element__(), prop.secondary)
525 else:
526 self._from_obj = ()
528 self._where_criteria = (
529 prop._with_parent(instance, alias_secondary=False),
530 )
532 if self.attr.order_by:
533 self._order_by_clauses = self.attr.order_by
534 else:
535 self._order_by_clauses = ()
537 def _add_all_impl(self, iterator: Iterable[_T]) -> None:
538 for item in iterator:
539 self.attr.append(
540 attributes.instance_state(self.instance),
541 attributes.instance_dict(self.instance),
542 item,
543 None,
544 )
546 def _remove_impl(self, item: _T) -> None:
547 self.attr.remove(
548 attributes.instance_state(self.instance),
549 attributes.instance_dict(self.instance),
550 item,
551 None,
552 )
555class WriteOnlyCollection(AbstractCollectionWriter[_T]):
556 """Write-only collection which can synchronize changes into the
557 attribute event system.
559 The :class:`.WriteOnlyCollection` is used in a mapping by
560 using the ``"write_only"`` lazy loading strategy with
561 :func:`_orm.relationship`. For background on this configuration,
562 see :ref:`write_only_relationship`.
564 .. versionadded:: 2.0
566 .. seealso::
568 :ref:`write_only_relationship`
570 """
572 __slots__ = (
573 "instance",
574 "attr",
575 "_where_criteria",
576 "_from_obj",
577 "_order_by_clauses",
578 )
580 def __iter__(self) -> NoReturn:
581 raise TypeError(
582 "WriteOnly collections don't support iteration in-place; "
583 "to query for collection items, use the select() method to "
584 "produce a SQL statement and execute it with session.scalars()."
585 )
587 def select(self) -> Select[_T]:
588 """Produce a :class:`_sql.Select` construct that represents the
589 rows within this instance-local :class:`_orm.WriteOnlyCollection`.
591 """
592 stmt = select(self.attr.target_mapper).where(*self._where_criteria)
593 if self._from_obj:
594 stmt = stmt.select_from(*self._from_obj)
595 if self._order_by_clauses:
596 stmt = stmt.order_by(*self._order_by_clauses)
597 return stmt
599 def insert(self) -> Insert:
600 """For one-to-many collections, produce a :class:`_dml.Insert` which
601 will insert new rows in terms of this this instance-local
602 :class:`_orm.WriteOnlyCollection`.
604 This construct is only supported for a :class:`_orm.Relationship`
605 that does **not** include the :paramref:`_orm.relationship.secondary`
606 parameter. For relationships that refer to a many-to-many table,
607 use ordinary bulk insert techniques to produce new objects, then
608 use :meth:`_orm.AbstractCollectionWriter.add_all` to associate them
609 with the collection.
612 """
614 state = inspect(self.instance)
615 mapper = state.mapper
616 prop = mapper._props[self.attr.key]
618 if prop.direction is not RelationshipDirection.ONETOMANY:
619 raise exc.InvalidRequestError(
620 "Write only bulk INSERT only supported for one-to-many "
621 "collections; for many-to-many, use a separate bulk "
622 "INSERT along with add_all()."
623 )
625 dict_: Dict[str, Any] = {}
627 for l, r in prop.synchronize_pairs:
628 fn = prop._get_attr_w_warn_on_none(
629 mapper,
630 state,
631 state.dict,
632 l,
633 )
635 dict_[r.key] = bindparam(None, callable_=fn)
637 return insert(self.attr.target_mapper).values(**dict_)
639 def update(self) -> Update:
640 """Produce a :class:`_dml.Update` which will refer to rows in terms
641 of this instance-local :class:`_orm.WriteOnlyCollection`.
643 """
644 return update(self.attr.target_mapper).where(*self._where_criteria)
646 def delete(self) -> Delete:
647 """Produce a :class:`_dml.Delete` which will refer to rows in terms
648 of this instance-local :class:`_orm.WriteOnlyCollection`.
650 """
651 return delete(self.attr.target_mapper).where(*self._where_criteria)
653 def add_all(self, iterator: Iterable[_T]) -> None:
654 """Add an iterable of items to this :class:`_orm.WriteOnlyCollection`.
656 The given items will be persisted to the database in terms of
657 the parent instance's collection on the next flush.
659 """
660 self._add_all_impl(iterator)
662 def add(self, item: _T) -> None:
663 """Add an item to this :class:`_orm.WriteOnlyCollection`.
665 The given item will be persisted to the database in terms of
666 the parent instance's collection on the next flush.
668 """
669 self._add_all_impl([item])
671 def remove(self, item: _T) -> None:
672 """Remove an item from this :class:`_orm.WriteOnlyCollection`.
674 The given item will be removed from the parent instance's collection on
675 the next flush.
677 """
678 self._remove_impl(item)