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

1from collections.abc import Mapping, Sequence 9 ctx1abcgdhefi

2from typing import ( 9 ctx1abcgdhefi

3 Any, 

4 TypeVar, 

5 cast, 

6 overload, 

7) 

8 

9from sqlalchemy import util 9 ctx1abcgdhefi

10from sqlalchemy.engine.cursor import CursorResult 9 ctx1abcgdhefi

11from sqlalchemy.engine.interfaces import _CoreAnyExecuteParams 9 ctx1abcgdhefi

12from sqlalchemy.engine.result import Result, ScalarResult, TupleResult 9 ctx1abcgdhefi

13from sqlalchemy.ext.asyncio import AsyncSession as _AsyncSession 9 ctx1abcgdhefi

14from sqlalchemy.ext.asyncio.result import _ensure_sync_result 9 ctx1abcgdhefi

15from sqlalchemy.ext.asyncio.session import _EXECUTE_OPTIONS 9 ctx1abcgdhefi

16from sqlalchemy.orm._typing import OrmExecuteOptionsParameter 9 ctx1abcgdhefi

17from sqlalchemy.sql.base import Executable as _Executable 9 ctx1abcgdhefi

18from sqlalchemy.sql.dml import UpdateBase 9 ctx1abcgdhefi

19from sqlalchemy.util.concurrency import greenlet_spawn 9 ctx1abcgdhefi

20from typing_extensions import TypeVarTuple, Unpack, deprecated 9 ctx1abcgdhefi

21 

22from ...orm.session import Session 9 ctx1abcgdhefi

23from ...sql.base import Executable 9 ctx1abcgdhefi

24from ...sql.expression import Select, SelectOfScalar 9 ctx1abcgdhefi

25 

26_TSelectParam = TypeVar("_TSelectParam", bound=Any) 9 ctx1abcgdhefi

27_Ts = TypeVarTuple("_Ts") 9 ctx1abcgdhefi

28 

29 

30class AsyncSession(_AsyncSession): 9 ctx1abcgdhefi

31 sync_session_class: type[Session] = Session 9 ctx1abcgdhefi

32 sync_session: Session 6 ctx1abcdef

33 

34 @overload 9 ctx1abcgdhefi

35 async def exec( 9 ctx1abcgdhefi

36 self, 

37 statement: Select[Unpack[_Ts]], 6 ctx1abcdef

38 *, 

39 params: Mapping[str, Any] | Sequence[Mapping[str, Any]] | None = None, 9 ctx1abcgdhefi

40 execution_options: Mapping[str, Any] = util.EMPTY_DICT, 9 ctx1abcgdhefi

41 bind_arguments: dict[str, Any] | None = None, 9 ctx1abcgdhefi

42 _parent_execute_state: Any | None = None, 9 ctx1abcgdhefi

43 _add_event: Any | None = None, 9 ctx1abcgdhefi

44 ) -> TupleResult[tuple[Unpack[_Ts]]]: ... 6 ctx1abcdef

45 

46 @overload 9 ctx1abcgdhefi

47 async def exec( 9 ctx1abcgdhefi

48 self, 

49 statement: SelectOfScalar[_TSelectParam], 6 ctx1abcdef

50 *, 

51 params: Mapping[str, Any] | Sequence[Mapping[str, Any]] | None = None, 9 ctx1abcgdhefi

52 execution_options: Mapping[str, Any] = util.EMPTY_DICT, 9 ctx1abcgdhefi

53 bind_arguments: dict[str, Any] | None = None, 9 ctx1abcgdhefi

54 _parent_execute_state: Any | None = None, 9 ctx1abcgdhefi

55 _add_event: Any | None = None, 9 ctx1abcgdhefi

56 ) -> ScalarResult[_TSelectParam]: ... 6 ctx1abcdef

57 

58 @overload 9 ctx1abcgdhefi

59 async def exec( 9 ctx1abcgdhefi

60 self, 

61 statement: UpdateBase, 6 ctx1abcdef

62 *, 

63 params: Mapping[str, Any] | Sequence[Mapping[str, Any]] | None = None, 9 ctx1abcgdhefi

64 execution_options: Mapping[str, Any] = util.EMPTY_DICT, 9 ctx1abcgdhefi

65 bind_arguments: dict[str, Any] | None = None, 9 ctx1abcgdhefi

66 _parent_execute_state: Any | None = None, 9 ctx1abcgdhefi

67 _add_event: Any | None = None, 9 ctx1abcgdhefi

68 ) -> CursorResult[Unpack[tuple[Any, ...]]]: ... 6 ctx1abcdef

69 

70 async def exec( 9 ctx1abcgdhefi

71 self, 

72 statement: Select[Unpack[_Ts]] 

73 | SelectOfScalar[_TSelectParam] 

74 | Executable[_TSelectParam] 

75 | UpdateBase, 

76 *, 

77 params: Mapping[str, Any] | Sequence[Mapping[str, Any]] | None = None, 

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

79 bind_arguments: dict[str, Any] | None = None, 

80 _parent_execute_state: Any | None = None, 

81 _add_event: Any | None = None, 

82 ) -> ( 

83 TupleResult[tuple[Unpack[_Ts]]] 

84 | ScalarResult[_TSelectParam] 

85 | CursorResult[Unpack[tuple[Any, ...]]] 

86 ): 

87 if execution_options: 9 ctx1jlmnkopqr

88 execution_options = util.immutabledict(execution_options).union( 9 ctx1jlmnkopqr

89 _EXECUTE_OPTIONS 

90 ) 

91 else: 

92 execution_options = _EXECUTE_OPTIONS 9 ctx1jlmnkopqr

93 

94 result = await greenlet_spawn( 9 ctx1jlmnkopqr

95 self.sync_session.exec, 

96 statement, 

97 params=params, 

98 execution_options=execution_options, 

99 bind_arguments=bind_arguments, 

100 _parent_execute_state=_parent_execute_state, 

101 _add_event=_add_event, 

102 ) 

103 result_value = await _ensure_sync_result( 9 ctx1jlmnkopqr

104 cast(Result[Unpack[tuple[Any, ...]]], result), self.exec 

105 ) 

106 return result_value # type: ignore 9 ctx1jlmnkopqr

107 

108 @deprecated( 11 ctx1ajbcgdkhefi

109 """ 

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

111 

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

113 of type `Row`, on which you have to call `scalars()` to get the model objects. 

114 

115 For example: 

116 

117 ```Python 

118 result = await session.execute(select(Hero)) 

119 heroes = result.scalars().all() 

120 ``` 

121 

122 Instead, you could use `exec()`: 

123 

124 ```Python 

125 result = await session.exec(select(Hero)) 

126 heroes = result.all() 

127 ``` 

128 """ 

129 ) 

130 async def execute( 9 ctx1abcgdhefi

131 self, 

132 statement: _Executable, 

133 params: _CoreAnyExecuteParams | None = None, 

134 *, 

135 execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT, 

136 bind_arguments: dict[str, Any] | None = None, 

137 _parent_execute_state: Any | None = None, 

138 _add_event: Any | None = None, 

139 ) -> Result[Unpack[tuple[Any, ...]]]: 

140 """ 

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

142 

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

144 of type `Row`, on which you have to call `scalars()` to get the model objects. 

145 

146 For example: 

147 

148 ```Python 

149 result = await session.execute(select(Hero)) 

150 heroes = result.scalars().all() 

151 ``` 

152 

153 Instead, you could use `exec()`: 

154 

155 ```Python 

156 result = await session.exec(select(Hero)) 

157 heroes = result.all() 

158 ``` 

159 """ 

160 return await super().execute( 9 ctx1jlmnkopqr

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 )