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.interfaces import _CoreAnyExecuteParams 

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

17from sqlalchemy.ext.asyncio import AsyncSession as _AsyncSession 

18from sqlalchemy.ext.asyncio.result import _ensure_sync_result 

19from sqlalchemy.ext.asyncio.session import _EXECUTE_OPTIONS 

20from sqlalchemy.orm._typing import OrmExecuteOptionsParameter 

21from sqlalchemy.sql.base import Executable as _Executable 

22from sqlalchemy.util.concurrency import greenlet_spawn 

23from typing_extensions import deprecated 

24 

25from ...orm.session import Session 

26from ...sql.base import Executable 

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

28 

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

30 

31 

32class AsyncSession(_AsyncSession): 

33 sync_session_class: Type[Session] = Session 

34 sync_session: Session 

35 

36 @overload 

37 async def exec( 

38 self, 

39 statement: Select[_TSelectParam], 

40 *, 

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

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

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

44 _parent_execute_state: Optional[Any] = None, 

45 _add_event: Optional[Any] = None, 

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

47 

48 @overload 

49 async def exec( 

50 self, 

51 statement: SelectOfScalar[_TSelectParam], 

52 *, 

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

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

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

56 _parent_execute_state: Optional[Any] = None, 

57 _add_event: Optional[Any] = None, 

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

59 

60 async def exec( 

61 self, 

62 statement: Union[ 

63 Select[_TSelectParam], 

64 SelectOfScalar[_TSelectParam], 

65 Executable[_TSelectParam], 

66 ], 

67 *, 

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

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

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

71 _parent_execute_state: Optional[Any] = None, 

72 _add_event: Optional[Any] = None, 

73 ) -> Union[TupleResult[_TSelectParam], ScalarResult[_TSelectParam]]: 

74 if execution_options: 

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

76 _EXECUTE_OPTIONS 

77 ) 

78 else: 

79 execution_options = _EXECUTE_OPTIONS 

80 

81 result = await greenlet_spawn( 

82 self.sync_session.exec, 

83 statement, 

84 params=params, 

85 execution_options=execution_options, 

86 bind_arguments=bind_arguments, 

87 _parent_execute_state=_parent_execute_state, 

88 _add_event=_add_event, 

89 ) 

90 result_value = await _ensure_sync_result( 

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

92 ) 

93 return result_value # type: ignore 

94 

95 @deprecated( 

96 """ 

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

98 

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

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

101 

102 For example: 

103 

104 ```Python 

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

106 ``` 

107 

108 instead you could use `exec()`: 

109 

110 ```Python 

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

112 ``` 

113 """ 

114 ) 

115 async def execute( # type: ignore 

116 self, 

117 statement: _Executable, 

118 params: Optional[_CoreAnyExecuteParams] = None, 

119 *, 

120 execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT, 

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

122 _parent_execute_state: Optional[Any] = None, 

123 _add_event: Optional[Any] = None, 

124 ) -> Result[Any]: 

125 """ 

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

127 

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

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

130 

131 For example: 

132 

133 ```Python 

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

135 ``` 

136 

137 instead you could use `exec()`: 

138 

139 ```Python 

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

141 ``` 

142 """ 

143 return await super().execute( 

144 statement, 

145 params=params, 

146 execution_options=execution_options, 

147 bind_arguments=bind_arguments, 

148 _parent_execute_state=_parent_execute_state, 

149 _add_event=_add_event, 

150 )