Current File : /opt/hc_python/lib/python3.12/site-packages/sqlalchemy/orm//dynamic.py
# 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})
