1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
|
# orm/_typing.py
# Copyright (C) 2022 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
from __future__ import annotations
import operator
from typing import Any
from typing import Dict
from typing import Mapping
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 ..engine.interfaces import _CoreKnownExecutionOptions
from ..sql import roles
from ..sql._orm_types import DMLStrategyArgument as DMLStrategyArgument
from ..sql._orm_types import (
SynchronizeSessionArgument as SynchronizeSessionArgument,
)
from ..sql._typing import _HasClauseElement
from ..sql.elements import ColumnElement
from ..util.typing import Protocol
from ..util.typing import TypeGuard
if TYPE_CHECKING:
from .attributes import AttributeImpl
from .attributes import CollectionAttributeImpl
from .attributes import HasCollectionAdapter
from .attributes import QueryableAttribute
from .base import PassiveFlag
from .decl_api import registry as _registry_type
from .interfaces import InspectionAttr
from .interfaces import MapperProperty
from .interfaces import ORMOption
from .interfaces import UserDefinedOption
from .mapper import Mapper
from .relationships import RelationshipProperty
from .state import InstanceState
from .util import AliasedClass
from .util import AliasedInsp
from ..sql._typing import _CE
from ..sql.base import ExecutableOption
_T = TypeVar("_T", bound=Any)
_T_co = TypeVar("_T_co", bound=Any, covariant=True)
# I would have preferred this were bound=object however it seems
# to not travel in all situations when defined in that way.
_O = TypeVar("_O", bound=Any)
"""The 'ORM mapped object' type.
"""
_OO = TypeVar("_OO", bound=object)
"""The 'ORM mapped object, that's definitely object' type.
"""
if TYPE_CHECKING:
_RegistryType = _registry_type
_InternalEntityType = Union["Mapper[_T]", "AliasedInsp[_T]"]
_ExternalEntityType = Union[Type[_T], "AliasedClass[_T]"]
_EntityType = Union[
Type[_T], "AliasedClass[_T]", "Mapper[_T]", "AliasedInsp[_T]"
]
_ClassDict = Mapping[str, Any]
_InstanceDict = Dict[str, Any]
_IdentityKeyType = Tuple[Type[_T], Tuple[Any, ...], Optional[Any]]
_ORMColumnExprArgument = Union[
ColumnElement[_T],
_HasClauseElement,
roles.ExpressionElementRole[_T],
]
_ORMCOLEXPR = TypeVar("_ORMCOLEXPR", bound=ColumnElement[Any])
class _OrmKnownExecutionOptions(_CoreKnownExecutionOptions, total=False):
populate_existing: bool
autoflush: bool
synchronize_session: SynchronizeSessionArgument
dml_strategy: DMLStrategyArgument
is_delete_using: bool
is_update_from: bool
OrmExecuteOptionsParameter = Union[
_OrmKnownExecutionOptions, Mapping[str, Any]
]
class _ORMAdapterProto(Protocol):
"""protocol for the :class:`.AliasedInsp._orm_adapt_element` method
which is a synonym for :class:`.AliasedInsp._adapt_element`.
"""
def __call__(self, obj: _CE, key: Optional[str] = None) -> _CE:
...
class _LoaderCallable(Protocol):
def __call__(self, state: InstanceState[Any], passive: PassiveFlag) -> Any:
...
def is_orm_option(
opt: ExecutableOption,
) -> TypeGuard[ORMOption]:
return not opt._is_core # type: ignore
def is_user_defined_option(
opt: ExecutableOption,
) -> TypeGuard[UserDefinedOption]:
return not opt._is_core and opt._is_user_defined # type: ignore
def is_composite_class(obj: Any) -> bool:
# inlining is_dataclass(obj)
return hasattr(obj, "__composite_values__") or hasattr(
obj, "__dataclass_fields__"
)
if TYPE_CHECKING:
def insp_is_mapper_property(obj: Any) -> TypeGuard[MapperProperty[Any]]:
...
def insp_is_mapper(obj: Any) -> TypeGuard[Mapper[Any]]:
...
def insp_is_aliased_class(obj: Any) -> TypeGuard[AliasedInsp[Any]]:
...
def insp_is_attribute(
obj: InspectionAttr,
) -> TypeGuard[QueryableAttribute[Any]]:
...
def attr_is_internal_proxy(
obj: InspectionAttr,
) -> TypeGuard[QueryableAttribute[Any]]:
...
def prop_is_relationship(
prop: MapperProperty[Any],
) -> TypeGuard[RelationshipProperty[Any]]:
...
def is_collection_impl(
impl: AttributeImpl,
) -> TypeGuard[CollectionAttributeImpl]:
...
def is_has_collection_adapter(
impl: AttributeImpl,
) -> TypeGuard[HasCollectionAdapter]:
...
else:
insp_is_mapper_property = operator.attrgetter("is_property")
insp_is_mapper = operator.attrgetter("is_mapper")
insp_is_aliased_class = operator.attrgetter("is_aliased_class")
insp_is_attribute = operator.attrgetter("is_attribute")
attr_is_internal_proxy = operator.attrgetter("_is_internal_proxy")
is_collection_impl = operator.attrgetter("collection")
prop_is_relationship = operator.attrgetter("_is_relationship")
is_has_collection_adapter = operator.attrgetter(
"_is_has_collection_adapter"
)
|
