Coverage for sqlmodel/ext/asyncio/session.py: 0%

1from typing import ( 

2 Any, 

3 Dict, 

4 Mapping, 

5 Optional, 

6 Sequence, 

7 Type, 

8 TypeVar, 

9 Union, 

10 cast, 

11 overload, 

12) 

13 

14from sqlalchemy import util 

15from sqlalchemy.engine.cursor import CursorResult 

16from sqlalchemy.engine.interfaces import _CoreAnyExecuteParams 

17from sqlalchemy.engine.result import Result, ScalarResult, TupleResult 

18from sqlalchemy.ext.asyncio import AsyncSession as _AsyncSession 

19from sqlalchemy.ext.asyncio.result import _ensure_sync_result 

20from sqlalchemy.ext.asyncio.session import _EXECUTE_OPTIONS 

21from sqlalchemy.orm._typing import OrmExecuteOptionsParameter 

22from sqlalchemy.sql.base import Executable as _Executable 

23from sqlalchemy.sql.dml import UpdateBase 

24from sqlalchemy.util.concurrency import greenlet_spawn 

25from typing_extensions import deprecated 

26 

27from ...orm.session import Session 

28from ...sql.base import Executable 

29from ...sql.expression import Select, SelectOfScalar 

30 

31_TSelectParam = TypeVar("_TSelectParam", bound=Any) 

32 

33 

34class AsyncSession(_AsyncSession): 

35 sync_session_class: Type[Session] = Session 

36 sync_session: Session 

37 

38 @overload 

39 async def exec( 

40 self, 

41 statement: Select[_TSelectParam], 

42 *, 

43 params: Optional[Union[Mapping[str, Any], Sequence[Mapping[str, Any]]]] = None, 

44 execution_options: Mapping[str, Any] = util.EMPTY_DICT, 

45 bind_arguments: Optional[Dict[str, Any]] = None, 

46 _parent_execute_state: Optional[Any] = None, 

47 _add_event: Optional[Any] = None, 

48 ) -> TupleResult[_TSelectParam]: ... 

49 

50 @overload 

51 async def exec( 

52 self, 

53 statement: SelectOfScalar[_TSelectParam], 

54 *, 

55 params: Optional[Union[Mapping[str, Any], Sequence[Mapping[str, Any]]]] = None, 

56 execution_options: Mapping[str, Any] = util.EMPTY_DICT, 

57 bind_arguments: Optional[Dict[str, Any]] = None, 

58 _parent_execute_state: Optional[Any] = None, 

59 _add_event: Optional[Any] = None, 

60 ) -> ScalarResult[_TSelectParam]: ... 

61 

62 @overload 

63 async def exec( 

64 self, 

65 statement: UpdateBase, 

66 *, 

67 params: Optional[Union[Mapping[str, Any], Sequence[Mapping[str, Any]]]] = None, 

68 execution_options: Mapping[str, Any] = util.EMPTY_DICT, 

69 bind_arguments: Optional[Dict[str, Any]] = None, 

70 _parent_execute_state: Optional[Any] = None, 

71 _add_event: Optional[Any] = None, 

72 ) -> CursorResult[Any]: ... 

73 

74 async def exec( 

75 self, 

76 statement: Union[ 

77 Select[_TSelectParam], 

78 SelectOfScalar[_TSelectParam], 

79 Executable[_TSelectParam], 

80 UpdateBase, 

81 ], 

82 *, 

83 params: Optional[Union[Mapping[str, Any], Sequence[Mapping[str, Any]]]] = None, 

84 execution_options: Mapping[str, Any] = util.EMPTY_DICT, 

85 bind_arguments: Optional[Dict[str, Any]] = None, 

86 _parent_execute_state: Optional[Any] = None, 

87 _add_event: Optional[Any] = None, 

88 ) -> Union[ 

89 TupleResult[_TSelectParam], ScalarResult[_TSelectParam], CursorResult[Any] 

90 ]: 

91 if execution_options: 

92 execution_options = util.immutabledict(execution_options).union( 

93 _EXECUTE_OPTIONS 

94 ) 

95 else: 

96 execution_options = _EXECUTE_OPTIONS 

97 

98 result = await greenlet_spawn( 

99 self.sync_session.exec, 

100 statement, 

101 params=params, 

102 execution_options=execution_options, 

103 bind_arguments=bind_arguments, 

104 _parent_execute_state=_parent_execute_state, 

105 _add_event=_add_event, 

106 ) 

107 result_value = await _ensure_sync_result( 

108 cast(Result[_TSelectParam], result), self.exec 

109 ) 

110 return result_value # type: ignore 

111 

112 @deprecated( 

113 """ 

114 🚨 You probably want to use `session.exec()` instead of `session.execute()`. 

115 

116 This is the original SQLAlchemy `session.execute()` method that returns objects 

117 of type `Row`, and that you have to call `scalars()` to get the model objects. 

118 

119 For example: 

120 

121 ```Python 

122 heroes = await session.execute(select(Hero)).scalars().all() 

123 ``` 

124 

125 instead you could use `exec()`: 

126 

127 ```Python 

128 heroes = await session.exec(select(Hero)).all() 

129 ``` 

130 """ 

131 ) 

132 async def execute( 

133 self, 

134 statement: _Executable, 

135 params: Optional[_CoreAnyExecuteParams] = None, 

136 *, 

137 execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT, 

138 bind_arguments: Optional[Dict[str, Any]] = None, 

139 _parent_execute_state: Optional[Any] = None, 

140 _add_event: Optional[Any] = None, 

141 ) -> Result[Any]: 

142 """ 

143 🚨 You probably want to use `session.exec()` instead of `session.execute()`. 

144 

145 This is the original SQLAlchemy `session.execute()` method that returns objects 

146 of type `Row`, and that you have to call `scalars()` to get the model objects. 

147 

148 For example: 

149 

150 ```Python 

151 heroes = await session.execute(select(Hero)).scalars().all() 

152 ``` 

153 

154 instead you could use `exec()`: 

155 

156 ```Python 

157 heroes = await session.exec(select(Hero)).all() 

158 ``` 

159 """ 

160 return await super().execute( 

161 statement, 

162 params=params, 

163 execution_options=execution_options, 

164 bind_arguments=bind_arguments, 

165 _parent_execute_state=_parent_execute_state, 

166 _add_event=_add_event, 

167 )