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

1from collections.abc import Mapping, Sequence 

2from typing import ( 

3 Any, 

4 Optional, 

5 TypeVar, 

6 Union, 

7 cast, 

8 overload, 

9) 

10 

11from sqlalchemy import util 

12from sqlalchemy.engine.cursor import CursorResult 

13from sqlalchemy.engine.interfaces import _CoreAnyExecuteParams 

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

15from sqlalchemy.ext.asyncio import AsyncSession as _AsyncSession 

16from sqlalchemy.ext.asyncio.result import _ensure_sync_result 

17from sqlalchemy.ext.asyncio.session import _EXECUTE_OPTIONS 

18from sqlalchemy.orm._typing import OrmExecuteOptionsParameter 

19from sqlalchemy.sql.base import Executable as _Executable 

20from sqlalchemy.sql.dml import UpdateBase 

21from sqlalchemy.util.concurrency import greenlet_spawn 

22from typing_extensions import deprecated 

23 

24from ...orm.session import Session 

25from ...sql.base import Executable 

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

27 

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

29 

30 

31class AsyncSession(_AsyncSession): 

32 sync_session_class: type[Session] = Session 

33 sync_session: Session 

34 

35 @overload 

36 async def exec( 

37 self, 

38 statement: Select[_TSelectParam], 

39 *, 

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

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

42 bind_arguments: Optional[dict[str, Any]] = None, 

43 _parent_execute_state: Optional[Any] = None, 

44 _add_event: Optional[Any] = None, 

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

46 

47 @overload 

48 async def exec( 

49 self, 

50 statement: SelectOfScalar[_TSelectParam], 

51 *, 

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

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

54 bind_arguments: Optional[dict[str, Any]] = None, 

55 _parent_execute_state: Optional[Any] = None, 

56 _add_event: Optional[Any] = None, 

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

58 

59 @overload 

60 async def exec( 

61 self, 

62 statement: UpdateBase, 

63 *, 

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

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

66 bind_arguments: Optional[dict[str, Any]] = None, 

67 _parent_execute_state: Optional[Any] = None, 

68 _add_event: Optional[Any] = None, 

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

70 

71 async def exec( 

72 self, 

73 statement: Union[ 

74 Select[_TSelectParam], 

75 SelectOfScalar[_TSelectParam], 

76 Executable[_TSelectParam], 

77 UpdateBase, 

78 ], 

79 *, 

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

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

82 bind_arguments: Optional[dict[str, Any]] = None, 

83 _parent_execute_state: Optional[Any] = None, 

84 _add_event: Optional[Any] = None, 

85 ) -> Union[ 

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

87 ]: 

88 if execution_options: 

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

90 _EXECUTE_OPTIONS 

91 ) 

92 else: 

93 execution_options = _EXECUTE_OPTIONS 

94 

95 result = await greenlet_spawn( 

96 self.sync_session.exec, 

97 statement, 

98 params=params, 

99 execution_options=execution_options, 

100 bind_arguments=bind_arguments, 

101 _parent_execute_state=_parent_execute_state, 

102 _add_event=_add_event, 

103 ) 

104 result_value = await _ensure_sync_result( 

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

106 ) 

107 return result_value # type: ignore 

108 

109 @deprecated( 

110 """ 

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

112 

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

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

115 

116 For example: 

117 

118 ```Python 

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

120 ``` 

121 

122 instead you could use `exec()`: 

123 

124 ```Python 

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

126 ``` 

127 """ 

128 ) 

129 async def execute( 

130 self, 

131 statement: _Executable, 

132 params: Optional[_CoreAnyExecuteParams] = None, 

133 *, 

134 execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT, 

135 bind_arguments: Optional[dict[str, Any]] = None, 

136 _parent_execute_state: Optional[Any] = None, 

137 _add_event: Optional[Any] = None, 

138 ) -> Result[Any]: 

139 """ 

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

141 

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

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

144 

145 For example: 

146 

147 ```Python 

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

149 ``` 

150 

151 instead you could use `exec()`: 

152 

153 ```Python 

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

155 ``` 

156 """ 

157 return await super().execute( 

158 statement, 

159 params=params, 

160 execution_options=execution_options, 

161 bind_arguments=bind_arguments, 

162 _parent_execute_state=_parent_execute_state, 

163 _add_event=_add_event, 

164 )