diff options
Diffstat (limited to '.venv/lib/python3.12/site-packages/sqlalchemy/orm/dynamic.py')
| -rw-r--r-- | .venv/lib/python3.12/site-packages/sqlalchemy/orm/dynamic.py | 300 |
1 files changed, 300 insertions, 0 deletions
diff --git a/.venv/lib/python3.12/site-packages/sqlalchemy/orm/dynamic.py b/.venv/lib/python3.12/site-packages/sqlalchemy/orm/dynamic.py new file mode 100644 index 00000000..3c81c396 --- /dev/null +++ b/.venv/lib/python3.12/site-packages/sqlalchemy/orm/dynamic.py @@ -0,0 +1,300 @@ +# orm/dynamic.py +# Copyright (C) 2005-2025 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 + + +"""Dynamic collection API. + +Dynamic collections act like Query() objects for read operations and support +basic add/delete mutation. + +.. legacy:: the "dynamic" loader is a legacy feature, superseded by the + "write_only" loader. + + +""" + +from __future__ import annotations + +from typing import Any +from typing import Iterable +from typing import Iterator +from typing import List +from typing import Optional +from typing import Tuple +from typing import Type +from typing import TYPE_CHECKING +from typing import TypeVar +from typing import Union + +from . import attributes +from . import exc as orm_exc +from . import relationships +from . import util as orm_util +from .base import PassiveFlag +from .query import Query +from .session import object_session +from .writeonly import AbstractCollectionWriter +from .writeonly import WriteOnlyAttributeImpl +from .writeonly import WriteOnlyHistory +from .writeonly import WriteOnlyLoader +from .. import util +from ..engine import result + + +if TYPE_CHECKING: + from . import QueryableAttribute + from .mapper import Mapper + from .relationships import _RelationshipOrderByArg + from .session import Session + from .state import InstanceState + from .util import AliasedClass + from ..event import _Dispatch + from ..sql.elements import ColumnElement + +_T = TypeVar("_T", bound=Any) + + +class DynamicCollectionHistory(WriteOnlyHistory[_T]): + def __init__( + self, + attr: DynamicAttributeImpl, + state: InstanceState[_T], + passive: PassiveFlag, + apply_to: Optional[DynamicCollectionHistory[_T]] = None, + ) -> None: + if apply_to: + coll = AppenderQuery(attr, state).autoflush(False) + self.unchanged_items = util.OrderedIdentitySet(coll) + self.added_items = apply_to.added_items + self.deleted_items = apply_to.deleted_items + self._reconcile_collection = True + else: + self.deleted_items = util.OrderedIdentitySet() + self.added_items = util.OrderedIdentitySet() + self.unchanged_items = util.OrderedIdentitySet() + self._reconcile_collection = False + + +class DynamicAttributeImpl(WriteOnlyAttributeImpl): + _supports_dynamic_iteration = True + collection_history_cls = DynamicCollectionHistory[Any] + query_class: Type[AppenderMixin[Any]] # type: ignore[assignment] + + def __init__( + self, + class_: Union[Type[Any], AliasedClass[Any]], + key: str, + dispatch: _Dispatch[QueryableAttribute[Any]], + target_mapper: Mapper[_T], + order_by: _RelationshipOrderByArg, + query_class: Optional[Type[AppenderMixin[_T]]] = None, + **kw: Any, + ) -> None: + attributes.AttributeImpl.__init__( + self, class_, key, None, dispatch, **kw + ) + self.target_mapper = target_mapper + if order_by: + self.order_by = tuple(order_by) + if not query_class: + self.query_class = AppenderQuery + elif AppenderMixin in query_class.mro(): + self.query_class = query_class + else: + self.query_class = mixin_user_query(query_class) + + +@relationships.RelationshipProperty.strategy_for(lazy="dynamic") +class DynaLoader(WriteOnlyLoader): + impl_class = DynamicAttributeImpl + + +class AppenderMixin(AbstractCollectionWriter[_T]): + """A mixin that expects to be mixing in a Query class with + AbstractAppender. + + + """ + + query_class: Optional[Type[Query[_T]]] = None + _order_by_clauses: Tuple[ColumnElement[Any], ...] + + def __init__( + self, attr: DynamicAttributeImpl, state: InstanceState[_T] + ) -> None: + Query.__init__( + self, # type: ignore[arg-type] + attr.target_mapper, + None, + ) + super().__init__(attr, state) + + @property + def session(self) -> Optional[Session]: + sess = object_session(self.instance) + if sess is not None and sess.autoflush and self.instance in sess: + sess.flush() + if not orm_util.has_identity(self.instance): + return None + else: + return sess + + @session.setter + def session(self, session: Session) -> None: + self.sess = session + + def _iter(self) -> Union[result.ScalarResult[_T], result.Result[_T]]: + sess = self.session + if sess is None: + state = attributes.instance_state(self.instance) + if state.detached: + util.warn( + "Instance %s is detached, dynamic relationship cannot " + "return a correct result. This warning will become " + "a DetachedInstanceError in a future release." + % (orm_util.state_str(state)) + ) + + return result.IteratorResult( + result.SimpleResultMetaData([self.attr.class_.__name__]), + iter( + self.attr._get_collection_history( + attributes.instance_state(self.instance), + PassiveFlag.PASSIVE_NO_INITIALIZE, + ).added_items + ), + _source_supports_scalars=True, + ).scalars() + else: + return self._generate(sess)._iter() + + if TYPE_CHECKING: + + def __iter__(self) -> Iterator[_T]: ... + + def __getitem__(self, index: Any) -> Union[_T, List[_T]]: + sess = self.session + if sess is None: + return self.attr._get_collection_history( + attributes.instance_state(self.instance), + PassiveFlag.PASSIVE_NO_INITIALIZE, + ).indexed(index) + else: + return self._generate(sess).__getitem__(index) # type: ignore[no-any-return] # noqa: E501 + + def count(self) -> int: + sess = self.session + if sess is None: + return len( + self.attr._get_collection_history( + attributes.instance_state(self.instance), + PassiveFlag.PASSIVE_NO_INITIALIZE, + ).added_items + ) + else: + return self._generate(sess).count() + + def _generate( + self, + sess: Optional[Session] = None, + ) -> Query[_T]: + # note we're returning an entirely new Query class instance + # here without any assignment capabilities; the class of this + # query is determined by the session. + instance = self.instance + if sess is None: + sess = object_session(instance) + if sess is None: + raise orm_exc.DetachedInstanceError( + "Parent instance %s is not bound to a Session, and no " + "contextual session is established; lazy load operation " + "of attribute '%s' cannot proceed" + % (orm_util.instance_str(instance), self.attr.key) + ) + + if self.query_class: + query = self.query_class(self.attr.target_mapper, session=sess) + else: + query = sess.query(self.attr.target_mapper) + + query._where_criteria = self._where_criteria + query._from_obj = self._from_obj + query._order_by_clauses = self._order_by_clauses + + return query + + def add_all(self, iterator: Iterable[_T]) -> None: + """Add an iterable of items to this :class:`_orm.AppenderQuery`. + + The given items will be persisted to the database in terms of + the parent instance's collection on the next flush. + + This method is provided to assist in delivering forwards-compatibility + with the :class:`_orm.WriteOnlyCollection` collection class. + + .. versionadded:: 2.0 + + """ + self._add_all_impl(iterator) + + def add(self, item: _T) -> None: + """Add an item to this :class:`_orm.AppenderQuery`. + + The given item will be persisted to the database in terms of + the parent instance's collection on the next flush. + + This method is provided to assist in delivering forwards-compatibility + with the :class:`_orm.WriteOnlyCollection` collection class. + + .. versionadded:: 2.0 + + """ + self._add_all_impl([item]) + + def extend(self, iterator: Iterable[_T]) -> None: + """Add an iterable of items to this :class:`_orm.AppenderQuery`. + + 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 append(self, item: _T) -> None: + """Append an item to this :class:`_orm.AppenderQuery`. + + 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.AppenderQuery`. + + The given item will be removed from the parent instance's collection on + the next flush. + + """ + self._remove_impl(item) + + +class AppenderQuery(AppenderMixin[_T], Query[_T]): # type: ignore[misc] + """A dynamic query that supports basic collection storage operations. + + Methods on :class:`.AppenderQuery` include all methods of + :class:`_orm.Query`, plus additional methods used for collection + persistence. + + + """ + + +def mixin_user_query(cls: Any) -> type[AppenderMixin[Any]]: + """Return a new class with AppenderQuery functionality layered over.""" + name = "Appender" + cls.__name__ + return type(name, (AppenderMixin, cls), {"query_class": cls}) |