codekingpro/portable-devtools

1# orm/dynamic.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
7
8
9"""Dynamic collection API.
10
11Dynamic collections act like Query() objects for read operations and support
12basic add/delete mutation.
13
14.. legacy:: the "dynamic" loader is a legacy feature, superseded by the
15 "write_only" loader.
16
17
18"""
19
20from __future__ import annotations
21
22from typing import Any
23from typing import Iterable
24from typing import Iterator
25from typing import List
26from typing import Optional
27from typing import overload
28from typing import Tuple
29from typing import Type
30from typing import TYPE_CHECKING
31from typing import TypeVar
32from typing import Union
33
34from . import attributes
35from . import exc as orm_exc
36from . import relationships
37from . import util as orm_util
38from .base import PassiveFlag
39from .query import Query
40from .session import object_session
41from .writeonly import AbstractCollectionWriter
42from .writeonly import WriteOnlyAttributeImpl
43from .writeonly import WriteOnlyHistory
44from .writeonly import WriteOnlyLoader
45from .. import util
46from ..engine import result
47
48
49if TYPE_CHECKING:
50    from . import QueryableAttribute
51    from .mapper import Mapper
52    from .relationships import _RelationshipOrderByArg
53    from .session import Session
54    from .state import InstanceState
55    from .util import AliasedClass
56    from ..event import _Dispatch
57    from ..sql.elements import ColumnElement
58
59_T = TypeVar("_T", bound=Any)
60
61
62class DynamicCollectionHistory(WriteOnlyHistory[_T]):
63    def __init__(
64        self,
65        attr: DynamicAttributeImpl,
66        state: InstanceState[_T],
67        passive: PassiveFlag,
68        apply_to: Optional[DynamicCollectionHistory[_T]] = None,
69    ) -> None:
70        if apply_to:
71            coll = AppenderQuery(attr, state).autoflush(False)
72            self.unchanged_items = util.OrderedIdentitySet(coll)
73            self.added_items = apply_to.added_items
74            self.deleted_items = apply_to.deleted_items
75            self._reconcile_collection = True
76        else:
77            self.deleted_items = util.OrderedIdentitySet()
78            self.added_items = util.OrderedIdentitySet()
79            self.unchanged_items = util.OrderedIdentitySet()
80            self._reconcile_collection = False
81
82
83class DynamicAttributeImpl(WriteOnlyAttributeImpl):
84    _supports_dynamic_iteration = True
85    collection_history_cls = DynamicCollectionHistory[Any]
86    query_class: Type[AppenderMixin[Any]]  # type: ignore[assignment]
87
88    def __init__(
89        self,
90        class_: Union[Type[Any], AliasedClass[Any]],
91        key: str,
92        dispatch: _Dispatch[QueryableAttribute[Any]],
93        target_mapper: Mapper[_T],
94        order_by: _RelationshipOrderByArg,
95        query_class: Optional[Type[AppenderMixin[_T]]] = None,
96        **kw: Any,
97    ) -> None:
98        attributes.AttributeImpl.__init__(
99            self, class_, key, None, dispatch, **kw
100        )
101        self.target_mapper = target_mapper
102        if order_by:
103            self.order_by = tuple(order_by)
104        if not query_class:
105            self.query_class = AppenderQuery
106        elif AppenderMixin in query_class.mro():
107            self.query_class = query_class
108        else:
109            self.query_class = mixin_user_query(query_class)
110
111
112@relationships.RelationshipProperty.strategy_for(lazy="dynamic")
113class DynaLoader(WriteOnlyLoader):
114    impl_class = DynamicAttributeImpl
115
116
117class AppenderMixin(AbstractCollectionWriter[_T]):
118    """A mixin that expects to be mixing in a Query class with
119    AbstractAppender.
120
121
122    """
123
124    query_class: Optional[Type[Query[_T]]] = None
125    _order_by_clauses: Tuple[ColumnElement[Any], ...]
126
127    def __init__(
128        self, attr: DynamicAttributeImpl, state: InstanceState[_T]
129    ) -> None:
130        Query.__init__(
131            self,  # type: ignore[arg-type]
132            attr.target_mapper,
133            None,
134        )
135        super().__init__(attr, state)
136
137    @property
138    def session(self) -> Optional[Session]:
139        sess = object_session(self.instance)
140        if sess is not None and sess.autoflush and self.instance in sess:
141            sess.flush()
142        if not orm_util.has_identity(self.instance):
143            return None
144        else:
145            return sess
146
147    @session.setter
148    def session(self, session: Session) -> None:
149        self.sess = session
150
151    def _iter(self) -> Union[result.ScalarResult[_T], result.Result[_T]]:
152        sess = self.session
153        if sess is None:
154            state = attributes.instance_state(self.instance)
155            if state.detached:
156                util.warn(
157                    "Instance %s is detached, dynamic relationship cannot "
158                    "return a correct result.   This warning will become "
159                    "a DetachedInstanceError in a future release."
160                    % (orm_util.state_str(state))
161                )
162
163            return result.IteratorResult(
164                result.SimpleResultMetaData([self.attr.class_.__name__]),
165                iter(
166                    self.attr._get_collection_history(
167                        attributes.instance_state(self.instance),
168                        PassiveFlag.PASSIVE_NO_INITIALIZE,
169                    ).added_items
170                ),
171                _source_supports_scalars=True,
172            ).scalars()
173        else:
174            return self._generate(sess)._iter()
175
176    if TYPE_CHECKING:
177
178        def __iter__(self) -> Iterator[_T]: ...
179
180    @overload
181    def __getitem__(self, index: int) -> _T: ...
182
183    @overload
184    def __getitem__(self, index: slice) -> List[_T]: ...
185
186    def __getitem__(self, index: Union[int, slice]) -> Union[_T, List[_T]]:
187        sess = self.session
188        if sess is None:
189            return self.attr._get_collection_history(
190                attributes.instance_state(self.instance),
191                PassiveFlag.PASSIVE_NO_INITIALIZE,
192            ).indexed(index)
193        else:
194            return self._generate(sess).__getitem__(index)
195
196    def count(self) -> int:
197        sess = self.session
198        if sess is None:
199            return len(
200                self.attr._get_collection_history(
201                    attributes.instance_state(self.instance),
202                    PassiveFlag.PASSIVE_NO_INITIALIZE,
203                ).added_items
204            )
205        else:
206            return self._generate(sess).count()
207
208    def _generate(
209        self,
210        sess: Optional[Session] = None,
211    ) -> Query[_T]:
212        # note we're returning an entirely new Query class instance
213        # here without any assignment capabilities; the class of this
214        # query is determined by the session.
215        instance = self.instance
216        if sess is None:
217            sess = object_session(instance)
218            if sess is None:
219                raise orm_exc.DetachedInstanceError(
220                    "Parent instance %s is not bound to a Session, and no "
221                    "contextual session is established; lazy load operation "
222                    "of attribute '%s' cannot proceed"
223                    % (orm_util.instance_str(instance), self.attr.key)
224                )
225
226        if self.query_class:
227            query = self.query_class(self.attr.target_mapper, session=sess)
228        else:
229            query = sess.query(self.attr.target_mapper)
230
231        query._where_criteria = self._where_criteria
232        query._from_obj = self._from_obj
233        query._order_by_clauses = self._order_by_clauses
234
235        return query
236
237    def add_all(self, iterator: Iterable[_T]) -> None:
238        """Add an iterable of items to this :class:`_orm.AppenderQuery`.
239
240        The given items will be persisted to the database in terms of
241        the parent instance's collection on the next flush.
242
243        This method is provided to assist in delivering forwards-compatibility
244        with the :class:`_orm.WriteOnlyCollection` collection class.
245
246        .. versionadded:: 2.0
247
248        """
249        self._add_all_impl(iterator)
250
251    def add(self, item: _T) -> None:
252        """Add an item to this :class:`_orm.AppenderQuery`.
253
254        The given item will be persisted to the database in terms of
255        the parent instance's collection on the next flush.
256
257        This method is provided to assist in delivering forwards-compatibility
258        with the :class:`_orm.WriteOnlyCollection` collection class.
259
260        .. versionadded:: 2.0
261
262        """
263        self._add_all_impl([item])
264
265    def extend(self, iterator: Iterable[_T]) -> None:
266        """Add an iterable of items to this :class:`_orm.AppenderQuery`.
267
268        The given items will be persisted to the database in terms of
269        the parent instance's collection on the next flush.
270
271        """
272        self._add_all_impl(iterator)
273
274    def append(self, item: _T) -> None:
275        """Append an item to this :class:`_orm.AppenderQuery`.
276
277        The given item will be persisted to the database in terms of
278        the parent instance's collection on the next flush.
279
280        """
281        self._add_all_impl([item])
282
283    def remove(self, item: _T) -> None:
284        """Remove an item from this :class:`_orm.AppenderQuery`.
285
286        The given item will be removed from the parent instance's collection on
287        the next flush.
288
289        """
290        self._remove_impl(item)
291
292
293class AppenderQuery(AppenderMixin[_T], Query[_T]):  # type: ignore[misc]
294    """A dynamic query that supports basic collection storage operations.
295
296    Methods on :class:`.AppenderQuery` include all methods of
297    :class:`_orm.Query`, plus additional methods used for collection
298    persistence.
299
300
301    """
302
303
304def mixin_user_query(cls: Any) -> type[AppenderMixin[Any]]:
305    """Return a new class with AppenderQuery functionality layered over."""
306    name = "Appender" + cls.__name__
307    return type(name, (AppenderMixin, cls), {"query_class": cls})
308