diff options
Diffstat (limited to 'lib/sqlalchemy/orm/dynamic.py')
-rw-r--r-- | lib/sqlalchemy/orm/dynamic.py | 80 |
1 files changed, 65 insertions, 15 deletions
diff --git a/lib/sqlalchemy/orm/dynamic.py b/lib/sqlalchemy/orm/dynamic.py index 1b4f573b5..084ba969f 100644 --- a/lib/sqlalchemy/orm/dynamic.py +++ b/lib/sqlalchemy/orm/dynamic.py @@ -16,6 +16,12 @@ basic add/delete mutation. from __future__ import annotations +from typing import Any +from typing import Optional +from typing import overload +from typing import TYPE_CHECKING +from typing import Union + from . import attributes from . import exc as orm_exc from . import interfaces @@ -23,17 +29,27 @@ from . import relationships from . import strategies from . import util as orm_util from .base import object_mapper +from .base import PassiveFlag from .query import Query from .session import object_session from .. import exc from .. import log from .. import util from ..engine import result +from ..util.typing import Literal + +if TYPE_CHECKING: + from ._typing import _InstanceDict + from .attributes import _AdaptedCollectionProtocol + from .attributes import AttributeEventToken + from .attributes import CollectionAdapter + from .base import LoaderCallableStatus + from .state import InstanceState @log.class_logger @relationships.Relationship.strategy_for(lazy="dynamic") -class DynaLoader(strategies.AbstractRelationshipLoader): +class DynaLoader(strategies.AbstractRelationshipLoader, log.Identified): def init_class_attribute(self, mapper): self.is_class_level = True if not self.uselist: @@ -106,13 +122,47 @@ class DynamicAttributeImpl( else: return self.query_class(self, state) + @overload def get_collection( self, - state, - dict_, - user_data=None, - passive=attributes.PASSIVE_NO_INITIALIZE, - ): + state: InstanceState[Any], + dict_: _InstanceDict, + user_data: Literal[None] = ..., + passive: Literal[PassiveFlag.PASSIVE_OFF] = ..., + ) -> CollectionAdapter: + ... + + @overload + def get_collection( + self, + state: InstanceState[Any], + dict_: _InstanceDict, + user_data: _AdaptedCollectionProtocol = ..., + passive: PassiveFlag = ..., + ) -> CollectionAdapter: + ... + + @overload + def get_collection( + self, + state: InstanceState[Any], + dict_: _InstanceDict, + user_data: Optional[_AdaptedCollectionProtocol] = ..., + passive: PassiveFlag = ..., + ) -> Union[ + Literal[LoaderCallableStatus.PASSIVE_NO_RESULT], CollectionAdapter + ]: + ... + + def get_collection( + self, + state: InstanceState[Any], + dict_: _InstanceDict, + user_data: Optional[_AdaptedCollectionProtocol] = None, + passive: PassiveFlag = PassiveFlag.PASSIVE_OFF, + ) -> Union[ + Literal[LoaderCallableStatus.PASSIVE_NO_RESULT], CollectionAdapter + ]: if not passive & attributes.SQL_OK: data = self._get_collection_history(state, passive).added_items else: @@ -170,15 +220,15 @@ class DynamicAttributeImpl( def set( self, - state, - dict_, - value, - initiator=None, - passive=attributes.PASSIVE_OFF, - check_old=None, - pop=False, - _adapt=True, - ): + state: InstanceState[Any], + dict_: _InstanceDict, + value: Any, + initiator: Optional[AttributeEventToken] = None, + passive: PassiveFlag = PassiveFlag.PASSIVE_OFF, + check_old: Any = None, + pop: bool = False, + _adapt: bool = True, + ) -> None: if initiator and initiator.parent_token is self.parent_token: return |