codekingpro/portable-devtools

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