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

1from collections.abc import Mapping, Sequence 

2from typing import ( 

3 Any, 

4 TypeVar, 

5 cast, 

6 overload, 

7) 

8 

9from sqlalchemy import util 

10from sqlalchemy.engine.cursor import CursorResult 

11from sqlalchemy.engine.interfaces import _CoreAnyExecuteParams 

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

13from sqlalchemy.ext.asyncio import AsyncSession as _AsyncSession 

14from sqlalchemy.ext.asyncio.result import _ensure_sync_result 

15from sqlalchemy.ext.asyncio.session import _EXECUTE_OPTIONS 

16from sqlalchemy.orm._typing import OrmExecuteOptionsParameter 

17from sqlalchemy.sql.base import Executable as _Executable 

18from sqlalchemy.sql.dml import UpdateBase 

19from sqlalchemy.util.concurrency import greenlet_spawn 

20from typing_extensions import deprecated 

21 

22from ...orm.session import Session 

23from ...sql.base import Executable 

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

25 

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

27 

28 

29class AsyncSession(_AsyncSession): 

30 sync_session_class: type[Session] = Session 

31 sync_session: Session 

32 

33 @overload 

34 async def exec( 

35 self, 

36 statement: Select[_TSelectParam], 

37 *, 

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

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

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

41 _parent_execute_state: Any | None = None, 

42 _add_event: Any | None = None, 

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

44 

45 @overload 

46 async def exec( 

47 self, 

48 statement: SelectOfScalar[_TSelectParam], 

49 *, 

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

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

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

53 _parent_execute_state: Any | None = None, 

54 _add_event: Any | None = None, 

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

56 

57 @overload 

58 async def exec( 

59 self, 

60 statement: UpdateBase, 

61 *, 

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

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

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

65 _parent_execute_state: Any | None = None, 

66 _add_event: Any | None = None, 

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

68 

69 async def exec( 

70 self, 

71 statement: Select[_TSelectParam] 

72 | SelectOfScalar[_TSelectParam] 

73 | Executable[_TSelectParam] 

74 | UpdateBase, 

75 *, 

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

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

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

79 _parent_execute_state: Any | None = None, 

80 _add_event: Any | None = None, 

81 ) -> TupleResult[_TSelectParam] | ScalarResult[_TSelectParam] | CursorResult[Any]: 

82 if execution_options: 

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

84 _EXECUTE_OPTIONS 

85 ) 

86 else: 

87 execution_options = _EXECUTE_OPTIONS 

88 

89 result = await greenlet_spawn( 

90 self.sync_session.exec, 

91 statement, 

92 params=params, 

93 execution_options=execution_options, 

94 bind_arguments=bind_arguments, 

95 _parent_execute_state=_parent_execute_state, 

96 _add_event=_add_event, 

97 ) 

98 result_value = await _ensure_sync_result( 

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

100 ) 

101 return result_value # type: ignore 

102 

103 @deprecated( 

104 """ 

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

106 

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

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

109 

110 For example: 

111 

112 ```Python 

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

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

115 ``` 

116 

117 instead you could use `exec()`: 

118 

119 ```Python 

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

121 heroes = result.all() 

122 ``` 

123 """ 

124 ) 

125 async def execute( 

126 self, 

127 statement: _Executable, 

128 params: _CoreAnyExecuteParams | None = None, 

129 *, 

130 execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT, 

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

132 _parent_execute_state: Any | None = None, 

133 _add_event: Any | None = None, 

134 ) -> Result[Any]: 

135 """ 

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

137 

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

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

140 

141 For example: 

142 

143 ```Python 

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

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

146 ``` 

147 

148 instead you could use `exec()`: 

149 

150 ```Python 

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

152 heroes = result.all() 

153 ``` 

154 """ 

155 return await super().execute( 

156 statement, 

157 params=params, 

158 execution_options=execution_options, 

159 bind_arguments=bind_arguments, 

160 _parent_execute_state=_parent_execute_state, 

161 _add_event=_add_event, 

162 )