ZerethShell — File Manager

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})

Vazgeç