Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/sqlalchemy/orm/_typing.py: 71%

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 Protocol 

16from typing import Tuple 

17from typing import Type 

18from typing import TYPE_CHECKING 

19from typing import TypeVar 

20from typing import Union 

21 

22from ..engine.interfaces import _CoreKnownExecutionOptions 

23from ..sql import roles 

24from ..sql._orm_types import DMLStrategyArgument as DMLStrategyArgument 

25from ..sql._orm_types import ( 

26 SynchronizeSessionArgument as SynchronizeSessionArgument, 

27) 

28from ..sql._typing import _HasClauseElement 

29from ..sql.elements import ColumnElement 

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 )